basicswap_miserver/tests/basicswap/extended/test_xmr_persistent.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2021-2022 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.

"""
export RESET_TEST=true
export TEST_PATH=/tmp/test_persistent
mkdir -p ${TEST_PATH}/bin/{particl,monero,bitcoin}
cp ~/tmp/particl-0.21.2.9-x86_64-linux-gnu_nousb.tar.gz ${TEST_PATH}/bin/particl
cp ~/tmp/bitcoin-22.0-x86_64-linux-gnu.tar.gz ${TEST_PATH}/bin/bitcoin
XMR_VERSION=0.17.3.2 cp ~/tmp/monero-linux-x64-v${XMR_VERSION}.tar.bz2 ${TEST_RELOAD_PATH}/bin/monero/monero-${XMR_VERSION}-x86_64-linux-gnu.tar.bz2
export PYTHONPATH=$(pwd)
python tests/basicswap/extended/test_xmr_persistent.py


"""

import os
import sys
import json
import time
import random
import shutil
import signal
import logging
import unittest
import threading
import multiprocessing
from unittest.mock import patch

from basicswap.rpc_xmr import (
    callrpc_xmr_na,
)
from basicswap.rpc import (
    callrpc,
)
from tests.basicswap.mnemonics import mnemonics as test_mnemonics
from tests.basicswap.common import (
    read_json_api,
    waitForServer,
)
from basicswap.contrib.rpcauth import generate_salt, password_to_hmac

import basicswap.config as cfg
import bin.basicswap_prepare as prepareSystem
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'))
PARTICL_PORT_BASE = int(os.getenv('PARTICL_PORT_BASE', '11938'))
BITCOIN_PORT_BASE = int(os.getenv('BITCOIN_PORT_BASE', '10938'))
RESET_TEST = make_boolean(os.getenv('RESET_TEST', 'false'))

XMR_BASE_P2P_PORT = 17792
XMR_BASE_RPC_PORT = 29798
XMR_BASE_WALLET_RPC_PORT = 29998

PORT_OFS = 1
UI_PORT = 12700 + PORT_OFS

BASE_PART_RPC_PORT = 19792
BASE_BTC_RPC_PORT = 19796

NUM_NODES = int(os.getenv('NUM_NODES', 3))
EXTRA_CONFIG_JSON = json.loads(os.getenv('EXTRA_CONFIG_JSON', '{}'))


logger = logging.getLogger()
logger.level = logging.DEBUG
if not len(logger.handlers):
    logger.addHandler(logging.StreamHandler(sys.stdout))


def recursive_update_dict(base, new_vals):
    for key, value in new_vals.items():
        if key in base and isinstance(value, dict):
            recursive_update_dict(base[key], value)
        else:
            base[key] = value


def callpartrpc(node_id, method, params=[], wallet=None, base_rpc_port=BASE_PART_RPC_PORT + PORT_OFS):
    auth = 'test_part_{0}:test_part_pwd_{0}'.format(node_id)
    return callrpc(base_rpc_port + node_id, auth, method, params, wallet)


def callbtcrpc(node_id, method, params=[], wallet=None, base_rpc_port=BASE_BTC_RPC_PORT + PORT_OFS):
    auth = 'test_btc_{0}:test_btc_pwd_{0}'.format(node_id)
    return callrpc(base_rpc_port + node_id, auth, method, params, wallet)


def updateThread(cls):
    while not cls.delay_event.is_set():
        try:
            if cls.btc_addr is not None:
                callbtcrpc(0, 'generatetoaddress', [1, cls.btc_addr])
        except Exception as e:
            print('updateThread error', str(e))
        cls.delay_event.wait(random.randrange(cls.update_min, cls.update_max))


def updateThreadXmr(cls):
    while not cls.delay_event.is_set():
        try:
            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})
        except Exception as e:
            print('updateThreadXmr error', str(e))
        cls.delay_event.wait(random.randrange(cls.xmr_update_min, cls.xmr_update_max))


class Test(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        super(Test, cls).setUpClass()

        cls.update_min = int(os.getenv('UPDATE_THREAD_MIN_WAIT', '1'))
        cls.update_max = cls.update_min * 4

        cls.xmr_update_min = int(os.getenv('XMR_UPDATE_THREAD_MIN_WAIT', '1'))
        cls.xmr_update_max = cls.xmr_update_min * 4

        cls.delay_event = threading.Event()
        cls.update_thread = None
        cls.update_thread_xmr = None
        cls.processes = []
        cls.btc_addr = None
        cls.xmr_addr = None

        random.seed(time.time())

        logging.info('Preparing %d nodes.', NUM_NODES)
        for i in range(NUM_NODES):
            logging.info('Preparing node: %d.', i)
            client_path = os.path.join(test_path, 'client{}'.format(i))
            config_path = os.path.join(client_path, cfg.CONFIG_FILENAME)
            if RESET_TEST:
                try:
                    logging.info('Removing dir %s', client_path)
                    shutil.rmtree(client_path)
                except Exception as ex:
                    logging.warning('setUpClass %s', str(ex))

            if not os.path.exists(config_path):

                os.environ['PART_RPC_PORT'] = str(BASE_PART_RPC_PORT)
                os.environ['BTC_RPC_PORT'] = str(BASE_BTC_RPC_PORT)

                testargs = [
                    'basicswap-prepare',
                    '-datadir="{}"'.format(client_path),
                    '-bindir="{}"'.format(os.path.join(test_path, 'bin')),
                    '-portoffset={}'.format(i + PORT_OFS),
                    '-regtest',
                    '-withcoins=monero,bitcoin',
                    '-noextractover',
                    '-xmrrestoreheight=0']
                if i < len(test_mnemonics):
                    testargs.append('-particl_mnemonic="{}"'.format(test_mnemonics[i]))
                with patch.object(sys, 'argv', testargs):
                    prepareSystem.main()

                with open(os.path.join(client_path, 'particl', 'particl.conf'), 'r') as fp:
                    lines = fp.readlines()
                with open(os.path.join(client_path, 'particl', 'particl.conf'), 'w') as fp:
                    for line in lines:
                        if not line.startswith('staking'):
                            fp.write(line)
                    fp.write('port={}\n'.format(PARTICL_PORT_BASE + i + PORT_OFS))
                    fp.write('bind=127.0.0.1\n')
                    fp.write('dnsseed=0\n')
                    fp.write('discover=0\n')
                    fp.write('listenonion=0\n')
                    fp.write('upnp=0\n')
                    fp.write('minstakeinterval=5\n')
                    fp.write('smsgsregtestadjust=0\n')
                    salt = generate_salt(16)
                    fp.write('rpcauth={}:{}${}\n'.format('test_part_' + str(i), salt, password_to_hmac(salt, 'test_part_pwd_' + str(i))))
                    for ip in range(NUM_NODES):
                        if ip != i:
                            fp.write('connect=127.0.0.1:{}\n'.format(PARTICL_PORT_BASE + ip + PORT_OFS))
                    for opt in EXTRA_CONFIG_JSON.get('part{}'.format(i), []):
                        fp.write(opt + '\n')

                # Pruned nodes don't provide blocks
                with open(os.path.join(client_path, 'bitcoin', 'bitcoin.conf'), 'r') as fp:
                    lines = fp.readlines()
                with open(os.path.join(client_path, 'bitcoin', 'bitcoin.conf'), 'w') as fp:
                    for line in lines:
                        if not line.startswith('prune'):
                            fp.write(line)
                    fp.write('port={}\n'.format(BITCOIN_PORT_BASE + i + PORT_OFS))
                    fp.write('bind=127.0.0.1\n')
                    fp.write('dnsseed=0\n')
                    fp.write('discover=0\n')
                    fp.write('listenonion=0\n')
                    fp.write('upnp=0\n')
                    salt = generate_salt(16)
                    fp.write('rpcauth={}:{}${}\n'.format('test_btc_' + str(i), salt, password_to_hmac(salt, 'test_btc_pwd_' + str(i))))
                    for ip in range(NUM_NODES):
                        if ip != i:
                            fp.write('connect=127.0.0.1:{}\n'.format(BITCOIN_PORT_BASE + ip + PORT_OFS))
                    for opt in EXTRA_CONFIG_JSON.get('btc{}'.format(i), []):
                        fp.write(opt + '\n')

                with open(os.path.join(client_path, 'monero', 'monerod.conf'), 'a') as fp:
                    fp.write('p2p-bind-ip=127.0.0.1\n')
                    fp.write('p2p-bind-port={}\n'.format(XMR_BASE_P2P_PORT + i + PORT_OFS))
                    for ip in range(NUM_NODES):
                        if ip != i:
                            fp.write('add-exclusive-node=127.0.0.1:{}\n'.format(XMR_BASE_P2P_PORT + ip + PORT_OFS))

                with open(config_path) as fs:
                    settings = json.load(fs)

                settings['min_delay_event'] = 1
                settings['max_delay_event'] = 4
                settings['min_delay_event_short'] = 1
                settings['max_delay_event_short'] = 4
                settings['min_delay_retry'] = 15
                settings['max_delay_retry'] = 30
                settings['min_sequence_lock_seconds'] = 60

                settings['check_progress_seconds'] = 5
                settings['check_watched_seconds'] = 5
                settings['check_expired_seconds'] = 60
                settings['check_events_seconds'] = 5
                settings['check_xmr_swaps_seconds'] = 5

                settings['chainclients']['particl']['rpcuser'] = 'test_part_' + str(i)
                settings['chainclients']['particl']['rpcpassword'] = 'test_part_pwd_' + str(i)

                settings['chainclients']['bitcoin']['rpcuser'] = 'test_btc_' + str(i)
                settings['chainclients']['bitcoin']['rpcpassword'] = 'test_btc_pwd_' + str(i)

                extra_config = EXTRA_CONFIG_JSON.get('sc{}'.format(i), {})
                recursive_update_dict(settings, extra_config)

                with open(config_path, 'w') as fp:
                    json.dump(settings, fp, indent=4)

        signal.signal(signal.SIGINT, lambda signal, frame: cls.signal_handler(cls, signal, frame))

    def signal_handler(self, sig, frame):
        logging.info('signal {} detected.'.format(sig))
        self.delay_event.set()

    def run_thread(self, client_id):
        client_path = os.path.join(test_path, 'client{}'.format(client_id))
        testargs = ['basicswap-run', '-datadir=' + client_path, '-regtest']
        with patch.object(sys, 'argv', testargs):
            runSystem.main()

    def start_processes(self):
        self.delay_event.clear()

        for i in range(NUM_NODES):
            self.processes.append(multiprocessing.Process(target=self.run_thread, args=(i,)))
            self.processes[-1].start()

        for i in range(NUM_NODES):
            waitForServer(self.delay_event, UI_PORT + i)

        wallets = read_json_api(UI_PORT + 1, 'wallets')

        self.xmr_addr = wallets['6']['main_address']
        num_blocks = 100
        if callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'] < num_blocks:
            logging.info('Mining {} Monero blocks to {}.'.format(num_blocks, self.xmr_addr))
            callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': self.xmr_addr, 'amount_of_blocks': num_blocks})
        logging.info('XMR blocks: %d', callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'])

        self.btc_addr = callbtcrpc(0, 'getnewaddress', ['mining_addr', 'bech32'])
        num_blocks = 500  # Mine enough to activate segwit
        if callbtcrpc(0, 'getblockchaininfo')['blocks'] < num_blocks:
            logging.info('Mining %d Bitcoin blocks to %s', num_blocks, self.btc_addr)
            callbtcrpc(0, 'generatetoaddress', [num_blocks, self.btc_addr])
        logging.info('BTC blocks: %d', callbtcrpc(0, 'getblockchaininfo')['blocks'])

        # Lower output split threshold for more stakeable outputs
        for i in range(NUM_NODES):
            callpartrpc(i, 'walletsettings', ['stakingoptions', {'stakecombinethreshold': 100, 'stakesplitthreshold': 200}])
        self.update_thread = threading.Thread(target=updateThread, args=(self,))
        self.update_thread.start()

        self.update_thread_xmr = threading.Thread(target=updateThreadXmr, args=(self,))
        self.update_thread_xmr.start()

        # Wait for height, or sequencelock is thrown off by genesis blocktime
        num_blocks = 3
        logging.info('Waiting for Particl chain height %d', num_blocks)
        for i in range(60):
            if self.delay_event.is_set():
                raise ValueError('Test stopped.')
            particl_blocks = callpartrpc(0, 'getblockchaininfo')['blocks']
            print('particl_blocks', particl_blocks)
            if particl_blocks >= num_blocks:
                break
            self.delay_event.wait(1)

        logging.info('PART blocks: %d', callpartrpc(0, 'getblockchaininfo')['blocks'])
        assert(particl_blocks >= num_blocks)

    @classmethod
    def tearDownClass(cls):
        logging.info('Stopping test')
        cls.delay_event.set()
        if cls.update_thread:
            cls.update_thread.join()
        if cls.update_thread_xmr:
            cls.update_thread_xmr.join()
        for p in cls.processes:
            p.terminate()
        for p in cls.processes:
            p.join()
        cls.update_thread = None
        cls.update_thread_xmr = None
        cls.processes = []

    def test_persistent(self):

        self.start_processes()

        waitForServer(self.delay_event, UI_PORT + 0)
        waitForServer(self.delay_event, UI_PORT + 1)

        while not self.delay_event.is_set():
            logging.info('Looping indefinitely, ctrl+c to exit.')
            self.delay_event.wait(10)


if __name__ == '__main__':
    unittest.main()