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

Add to test_xmr_persistent.

Browse source
This commit is contained in:
tecnovert 2024-12-01 17:11:44 +02:00
parent 889ffaaa33
commit 3be72b3c71
8 changed files with 397 additions and 216 deletions

2
basicswap/basicswap.py
View file

@ -355,7 +355,7 @@ class BasicSwap(BaseApp):
# TODO: Set dynamically
self.balance_only_coins = (Coins.LTC_MWEB,)
self.scriptless_coins = (Coins.XMR, Coins.WOW, Coins.PART_ANON, Coins.FIRO)
self.scriptless_coins = (Coins.XMR, Coins.WOW, Coins.PART_ANON, Coins.FIRO, Coins.DOGE)
self.adaptor_swap_only_coins = self.scriptless_coins + (
Coins.PART_BLIND,
Coins.BCH,

50
basicswap/bin/prepare.py
View file

@ -53,15 +53,9 @@ PARTICL_LINUX_EXTRA = os.getenv("PARTICL_LINUX_EXTRA", "nousb")
LITECOIN_VERSION = os.getenv("LITECOIN_VERSION", "0.21.3")
LITECOIN_VERSION_TAG = os.getenv("LITECOIN_VERSION_TAG", "")
DOGECOIN_VERSION = os.getenv("DOGECOIN_VERSION", "1.14.7")
DOGECOIN_VERSION_TAG = os.getenv("DOGECOIN_VERSION_TAG", "")
BITCOIN_VERSION = os.getenv("BITCOIN_VERSION", "26.0")
BITCOIN_VERSION_TAG = os.getenv("BITCOIN_VERSION_TAG", "")
BITCOINCASH_VERSION = os.getenv("BITCOINCASH_VERSION", "27.1.0")
BITCOINCASH_VERSION_TAG = os.getenv("BITCOINCASH_VERSION_TAG", "")
MONERO_VERSION = os.getenv("MONERO_VERSION", "0.18.3.4")
MONERO_VERSION_TAG = os.getenv("MONERO_VERSION_TAG", "")
XMR_SITE_COMMIT = (
@ -89,6 +83,12 @@ NAV_VERSION_TAG = os.getenv("NAV_VERSION_TAG", "")
DCR_VERSION = os.getenv("DCR_VERSION", "1.8.1")
DCR_VERSION_TAG = os.getenv("DCR_VERSION_TAG", "")
BITCOINCASH_VERSION = os.getenv("BITCOINCASH_VERSION", "27.1.0")
BITCOINCASH_VERSION_TAG = os.getenv("BITCOINCASH_VERSION_TAG", "")
DOGECOIN_VERSION = os.getenv("DOGECOIN_VERSION", "1.14.7")
DOGECOIN_VERSION_TAG = os.getenv("DOGECOIN_VERSION_TAG", "")
GUIX_SSL_CERT_DIR = None
ADD_PUBKEY_URL = os.getenv("ADD_PUBKEY_URL", "")
@ -101,9 +101,7 @@ SKIP_GPG_VALIDATION = toBool(os.getenv("SKIP_GPG_VALIDATION", "false"))
known_coins = {
"particl": (PARTICL_VERSION, PARTICL_VERSION_TAG, ("tecnovert",)),
"bitcoin": (BITCOIN_VERSION, BITCOIN_VERSION_TAG, ("laanwj",)),
"bitcoincash": (BITCOINCASH_VERSION, BITCOINCASH_VERSION_TAG, ("Calin_Culianu",)),
"litecoin": (LITECOIN_VERSION, LITECOIN_VERSION_TAG, ("davidburkett38",)),
"dogecoin": (DOGECOIN_VERSION, DOGECOIN_VERSION_TAG, ("patricklodder",)),
"decred": (DCR_VERSION, DCR_VERSION_TAG, ("decred_release",)),
"namecoin": ("0.18.0", "", ("JeremyRand",)),
"monero": (MONERO_VERSION, MONERO_VERSION_TAG, ("binaryfate",)),
@ -112,6 +110,8 @@ known_coins = {
"dash": (DASH_VERSION, DASH_VERSION_TAG, ("pasta",)),
"firo": (FIRO_VERSION, FIRO_VERSION_TAG, ("reuben",)),
"navcoin": (NAV_VERSION, NAV_VERSION_TAG, ("nav_builder",)),
"bitcoincash": (BITCOINCASH_VERSION, BITCOINCASH_VERSION_TAG, ("Calin_Culianu",)),
"dogecoin": (DOGECOIN_VERSION, DOGECOIN_VERSION_TAG, ("patricklodder",)),
}
disabled_coins = [
@ -203,25 +203,12 @@ LTC_ONION_PORT = int(os.getenv("LTC_ONION_PORT", 9333))
LTC_RPC_USER = os.getenv("LTC_RPC_USER", "")
LTC_RPC_PWD = os.getenv("LTC_RPC_PWD", "")
DOGE_RPC_HOST = os.getenv("DOGE_RPC_HOST", "127.0.0.1")
DOGE_RPC_PORT = int(os.getenv("DOGE_RPC_PORT", 42069))
DOGE_ONION_PORT = int(os.getenv("DOGE_ONION_PORT", 6969))
DOGE_RPC_USER = os.getenv("DOGE_RPC_USER", "")
DOGE_RPC_PWD = os.getenv("DOGE_RPC_PWD", "")
BTC_RPC_HOST = os.getenv("BTC_RPC_HOST", "127.0.0.1")
BTC_RPC_PORT = int(os.getenv("BTC_RPC_PORT", 19996))
BTC_ONION_PORT = int(os.getenv("BTC_ONION_PORT", 8334))
BTC_RPC_USER = os.getenv("BTC_RPC_USER", "")
BTC_RPC_PWD = os.getenv("BTC_RPC_PWD", "")
BCH_RPC_HOST = os.getenv("BCH_RPC_HOST", "127.0.0.1")
BCH_RPC_PORT = int(os.getenv("BCH_RPC_PORT", 19997))
BCH_ONION_PORT = int(os.getenv("BCH_ONION_PORT", 8335))
BCH_PORT = int(os.getenv("BCH_PORT", 19798))
BCH_RPC_USER = os.getenv("BCH_RPC_USER", "")
BCH_RPC_PWD = os.getenv("BCH_RPC_PWD", "")
DCR_RPC_HOST = os.getenv("DCR_RPC_HOST", "127.0.0.1")
DCR_RPC_PORT = int(os.getenv("DCR_RPC_PORT", 9109))
DCR_WALLET_RPC_HOST = os.getenv("DCR_WALLET_RPC_HOST", "127.0.0.1")
@ -259,6 +246,19 @@ NAV_ONION_PORT = int(os.getenv("NAV_ONION_PORT", 8334)) # TODO?
NAV_RPC_USER = os.getenv("NAV_RPC_USER", "")
NAV_RPC_PWD = os.getenv("NAV_RPC_PWD", "")
BCH_RPC_HOST = os.getenv("BCH_RPC_HOST", "127.0.0.1")
BCH_RPC_PORT = int(os.getenv("BCH_RPC_PORT", 19997))
BCH_ONION_PORT = int(os.getenv("BCH_ONION_PORT", 8335))
BCH_PORT = int(os.getenv("BCH_PORT", 19798))
BCH_RPC_USER = os.getenv("BCH_RPC_USER", "")
BCH_RPC_PWD = os.getenv("BCH_RPC_PWD", "")
DOGE_RPC_HOST = os.getenv("DOGE_RPC_HOST", "127.0.0.1")
DOGE_RPC_PORT = int(os.getenv("DOGE_RPC_PORT", 42069))
DOGE_ONION_PORT = int(os.getenv("DOGE_ONION_PORT", 6969))
DOGE_RPC_USER = os.getenv("DOGE_RPC_USER", "")
DOGE_RPC_PWD = os.getenv("DOGE_RPC_PWD", "")
TOR_PROXY_HOST = os.getenv("TOR_PROXY_HOST", "127.0.0.1")
TOR_PROXY_PORT = int(os.getenv("TOR_PROXY_PORT", 9050))
TOR_CONTROL_PORT = int(os.getenv("TOR_CONTROL_PORT", 9051))
@ -1250,7 +1250,7 @@ def prepareDataDir(coin, settings, chain, particl_mnemonic, extra_opts={}):
fp.write(chainname + "=1\n")
else:
fp.write(chain + "=1\n")
if coin not in ("firo", "navcoin"):
if coin not in ("firo", "navcoin", "dogecoin"):
if chain == "testnet":
fp.write("[test]\n\n")
elif chain == "regtest":
@ -1730,7 +1730,6 @@ def initialise_wallets(
Coins.PART,
Coins.BTC,
Coins.LTC,
Coins.DOGE,
Coins.DCR,
Coins.DASH,
)
@ -2298,10 +2297,11 @@ def main():
"onionport": DOGE_ONION_PORT + port_offset,
"datadir": os.getenv("DOGE_DATA_DIR", os.path.join(data_dir, "dogecoin")),
"bindir": os.path.join(bin_dir, "dogecoin"),
"use_segwit": True,
"use_segwit": False,
"use_csv": False,
"blocks_confirmed": 2,
"conf_target": 2,
"core_version_group": 21, # TODO ofrnxmr
"core_version_group": 14,
"min_relay_fee": 0.00001,
},
"decred": {

1
basicswap/chainparams.py
View file

@ -1,6 +1,7 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2019-2024 tecnovert
# Copyright (c) 2024 The Basicswap developers
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.

29
basicswap/interface/doge.py
View file

@ -1,16 +1,43 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2020-2024 tecnovert
# Copyright (c) 2024 The BasicSwap developers
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
from .btc import BTCInterface
from basicswap.chainparams import Coins
from basicswap.rpc import make_rpc_func
class DOGEInterface(BTCInterface):
@staticmethod
def coin_type():
return Coins.DOGE
def __init__(self, coin_settings, network, swap_client=None):
super(DOGEInterface, self).__init__(coin_settings, network, swap_client)
# No multiwallet support
self.rpc_wallet = make_rpc_func(
self._rpcport, self._rpcauth, host=self._rpc_host
)
def initialiseWallet(self, key):
# load with -hdseed= parameter
pass
def checkWallets(self) -> int:
return 1
def getNewAddress(self, use_segwit, label="swap_receive"):
return self.rpc("getnewaddress", [label])
def isWatchOnlyAddress(self, address):
addr_info = self.rpc("validateaddress", [address])
return addr_info["iswatchonly"]
def isAddressMine(self, address: str, or_watch_only: bool = False) -> bool:
addr_info = self.rpc("validateaddress", [address])
if not or_watch_only:
return addr_info["ismine"]
return addr_info["ismine"] or addr_info["iswatchonly"]

61
tests/basicswap/common_xmr.py
View file

@ -50,6 +50,17 @@ from basicswap.contrib.rpcauth import generate_salt, password_to_hmac
import basicswap.config as cfg
import basicswap.bin.run as runSystem
XMR_BASE_P2P_PORT = 17792
XMR_BASE_RPC_PORT = 29798
XMR_BASE_WALLET_RPC_PORT = 29998
FIRO_BASE_PORT = 34832
FIRO_BASE_RPC_PORT = 35832
FIRO_RPC_PORT_BASE = int(os.getenv("FIRO_RPC_PORT_BASE", FIRO_BASE_RPC_PORT))
DOGE_BASE_PORT = 22556
DOGE_BASE_RPC_PORT = 18442
TEST_PATH = os.path.expanduser(os.getenv("TEST_PATH", "~/test_basicswap1"))
PARTICL_PORT_BASE = int(os.getenv("PARTICL_PORT_BASE", BASE_PORT))
@ -64,16 +75,9 @@ DECRED_RPC_PORT_BASE = int(os.getenv("DECRED_RPC_PORT_BASE", DCR_BASE_RPC_PORT))
BITCOINCASH_RPC_PORT_BASE = int(
os.getenv("BITCOINCASH_RPC_PORT_BASE", BCH_BASE_RPC_PORT)
)
FIRO_BASE_PORT = 34832
FIRO_BASE_RPC_PORT = 35832
FIRO_RPC_PORT_BASE = int(os.getenv("FIRO_RPC_PORT_BASE", FIRO_BASE_RPC_PORT))
XMR_BASE_P2P_PORT = 17792
XMR_BASE_RPC_PORT = 29798
XMR_BASE_WALLET_RPC_PORT = 29998
DOGECOIN_RPC_PORT_BASE = int(
os.getenv("DOGECOIN_RPC_PORT_BASE", DOGE_BASE_RPC_PORT)
)
EXTRA_CONFIG_JSON = json.loads(os.getenv("EXTRA_CONFIG_JSON", "{}"))
@ -131,9 +135,11 @@ def run_prepare(
os.environ["BTC_RPC_PORT"] = str(BITCOIN_RPC_PORT_BASE)
os.environ["LTC_RPC_PORT"] = str(LITECOIN_RPC_PORT_BASE)
os.environ["DCR_RPC_PORT"] = str(DECRED_RPC_PORT_BASE)
os.environ["FIRO_RPC_PORT"] = str(FIRO_RPC_PORT_BASE)
os.environ["BCH_PORT"] = str(BCH_BASE_PORT)
os.environ["BCH_RPC_PORT"] = str(BITCOINCASH_RPC_PORT_BASE)
os.environ["FIRO_RPC_PORT"] = str(FIRO_RPC_PORT_BASE)
os.environ["DOGE_PORT"] = str(DOGE_BASE_PORT)
os.environ["DOGE_RPC_PORT"] = str(DOGECOIN_RPC_PORT_BASE)
os.environ["XMR_RPC_USER"] = "xmr_user"
os.environ["XMR_RPC_PWD"] = "xmr_pwd"
@ -433,6 +439,39 @@ def run_prepare(
for opt in EXTRA_CONFIG_JSON.get("bch{}".format(node_id), []):
fp.write(opt + "\n")
if "dogecoin" in coins_array:
config_filename = os.path.join(datadir_path, "dogecoin", "dogecoin.conf")
with open(config_filename, "r") as fp:
lines = fp.readlines()
with open(config_filename, "w") as fp:
for line in lines:
if not line.startswith("prune"):
fp.write(line)
fp.write("port={}\n".format(DOGE_BASE_PORT + node_id + 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")
if use_rpcauth:
salt = generate_salt(16)
rpc_user = "test_doge_" + str(node_id)
rpc_pass = "test_doge_pwd_" + str(node_id)
fp.write(
"rpcauth={}:{}${}\n".format(
rpc_user, salt, password_to_hmac(salt, rpc_pass)
)
)
settings["chainclients"]["dogecoin"]["rpcuser"] = rpc_user
settings["chainclients"]["dogecoin"]["rpcpassword"] = rpc_pass
for ip in range(num_nodes):
if ip != node_id:
fp.write(
"connect=127.0.0.1:{}\n".format(DOGE_BASE_PORT + ip + port_ofs)
)
for opt in EXTRA_CONFIG_JSON.get("doge{}".format(node_id), []):
fp.write(opt + "\n")
with open(config_path) as fs:
settings = json.load(fs)

69
tests/basicswap/extended/test_doge.py Normal file
View file

@ -0,0 +1,69 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2024 The Basicswap developers
# 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_doge
mkdir -p ${TEST_PATH}/bin
cp -r ~/tmp/basicswap_bin/* ${TEST_PATH}/bin
export PYTHONPATH=$(pwd)
export TEST_COINS_LIST='bitcoin,dogecoin'
python tests/basicswap/extended/test_doge.py
"""
import os
import sys
import json
import time
import random
import signal
import logging
import unittest
import threading
import multiprocessing
from unittest.mock import patch
from tests.basicswap.extended.test_xmr_persistent import (
BaseTestWithPrepare,
UI_PORT,
)
from tests.basicswap.extended.test_scripts import (
wait_for_offers,
)
from tests.basicswap.util import (
read_json_api,
)
logger = logging.getLogger()
logger.level = logging.DEBUG
if not len(logger.handlers):
logger.addHandler(logging.StreamHandler(sys.stdout))
class DOGETest(BaseTestWithPrepare):
def test_a(self):
read_json_api(UI_PORT + 0, "wallets/doge/reseed")
read_json_api(UI_PORT + 1, "wallets/doge/reseed")
offer_json = {
"coin_from": "btc",
"coin_to": "doge",
"amt_from": 10.0,
"amt_to": 100.0,
"amt_var": True,
"lockseconds": 3600,
}
offer_id = read_json_api(UI_PORT + 0, "offers/new", offer_json)["offer_id"]
logging.debug(f"offer_id {offer_id}")
wait_for_offers(self.delay_event, 1, 1, offer_id)
if __name__ == "__main__":
unittest.main()

1
tests/basicswap/extended/test_wow.py
View file

@ -103,7 +103,6 @@ class Test(BaseTest):
@classmethod
def prepareExtraCoins(cls):
pass
num_blocks = 300
cls.wow_addr = cls.callwownodewallet(cls, 1, "get_address")["address"]
if callrpc_xmr(WOW_BASE_RPC_PORT + 1, "get_block_count")["count"] < num_blocks:

400
tests/basicswap/extended/test_xmr_persistent.py
View file

@ -56,11 +56,11 @@ from tests.basicswap.util import (
from tests.basicswap.common_xmr import (
prepare_nodes,
XMR_BASE_RPC_PORT,
DOGE_BASE_RPC_PORT,
)
from basicswap.interface.dcr.rpc import callrpc as callrpc_dcr
import basicswap.bin.run as runSystem
test_path = os.path.expanduser(os.getenv("TEST_PATH", "/tmp/test_persistent"))
RESET_TEST = make_boolean(os.getenv("RESET_TEST", "false"))
@ -70,6 +70,7 @@ UI_PORT = 12700 + PORT_OFS
PARTICL_RPC_PORT_BASE = int(os.getenv("PARTICL_RPC_PORT_BASE", BASE_RPC_PORT))
BITCOIN_RPC_PORT_BASE = int(os.getenv("BITCOIN_RPC_PORT_BASE", BTC_BASE_RPC_PORT))
LITECOIN_RPC_PORT_BASE = int(os.getenv("LITECOIN_RPC_PORT_BASE", LTC_BASE_RPC_PORT))
DOGECOIN_RPC_PORT_BASE = int(os.getenv("DOGECOIN_RPC_PORT_BASE", DOGE_BASE_RPC_PORT))
BITCOINCASH_RPC_PORT_BASE = int(
os.getenv("BITCOINCASH_RPC_PORT_BASE", BCH_BASE_RPC_PORT)
)
@ -137,6 +138,17 @@ def callbchrpc(
return callrpc(base_rpc_port + node_id, auth, method, params, wallet)
def calldogerpc(
node_id,
method,
params=[],
wallet=None,
base_rpc_port=DOGECOIN_RPC_PORT_BASE + PORT_OFS,
):
auth = "test_doge_{0}:test_doge_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:
@ -146,6 +158,8 @@ def updateThread(cls):
callltcrpc(0, "generatetoaddress", [1, cls.ltc_addr])
if cls.bch_addr is not None:
callbchrpc(0, "generatetoaddress", [1, cls.bch_addr])
if cls.doge_addr is not None:
calldogerpc(0, "generatetoaddress", [1, cls.doge_addr])
except Exception as e:
print("updateThread error", str(e))
cls.delay_event.wait(random.randrange(cls.update_min, cls.update_max))
@ -204,74 +218,33 @@ def updateThreadDCR(cls):
cls.delay_event.wait(random.randrange(cls.dcr_update_min, cls.dcr_update_max))
class Test(unittest.TestCase):
@classmethod
def setUpClass(cls):
super(Test, cls).setUpClass()
def signal_handler(self, sig, frame):
logging.info("signal {} detected.".format(sig))
self.delay_event.set()
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
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()
cls.dcr_update_min = int(os.getenv("DCR_UPDATE_THREAD_MIN_WAIT", "1"))
cls.dcr_update_max = cls.dcr_update_min * 4
cls.delay_event = threading.Event()
cls.update_thread = None
cls.update_thread_xmr = None
cls.update_thread_dcr = None
cls.processes = []
cls.btc_addr = None
cls.ltc_addr = None
cls.bch_addr = None
cls.xmr_addr = None
cls.dcr_addr = "SsYbXyjkKAEXXcGdFgr4u4bo4L8RkCxwQpH"
cls.dcr_acc = None
def start_processes(self):
self.delay_event.clear()
random.seed(time.time())
if os.path.exists(test_path) and not RESET_TEST:
logging.info(f"Continuing with existing directory: {test_path}")
else:
logging.info("Preparing %d nodes.", NUM_NODES)
prepare_nodes(
NUM_NODES,
TEST_COINS_LIST,
True,
{"min_sequence_lock_seconds": 60},
PORT_OFS,
)
signal.signal(
signal.SIGINT, lambda signal, frame: cls.signal_handler(cls, signal, frame)
for i in range(NUM_NODES):
self.processes.append(
multiprocessing.Process(target=run_thread, args=(self, i,))
)
self.processes[-1].start()
def signal_handler(self, sig, frame):
logging.info("signal {} detected.".format(sig))
self.delay_event.set()
for i in range(NUM_NODES):
waitForServer(self.delay_event, UI_PORT + i)
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")
wallets = read_json_api(UI_PORT + 1, "wallets")
if "monero" in TEST_COINS_LIST:
xmr_auth = None
if os.getenv("XMR_RPC_USER", "") != "":
xmr_auth = (os.getenv("XMR_RPC_USER", ""), os.getenv("XMR_RPC_PWD", ""))
@ -300,127 +273,192 @@ class Test(unittest.TestCase):
],
)
self.btc_addr = callbtcrpc(0, "getnewaddress", ["mining_addr", "bech32"])
num_blocks: int = 500 # Mine enough to activate segwit
if callbtcrpc(0, "getblockcount") < 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, "getblockcount"))
self.btc_addr = callbtcrpc(0, "getnewaddress", ["mining_addr", "bech32"])
num_blocks: int = 500 # Mine enough to activate segwit
if callbtcrpc(0, "getblockcount") < 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, "getblockcount"))
if "litecoin" in TEST_COINS_LIST:
self.ltc_addr = callltcrpc(
0, "getnewaddress", ["mining_addr"], wallet="wallet.dat"
if "litecoin" in TEST_COINS_LIST:
self.ltc_addr = callltcrpc(
0, "getnewaddress", ["mining_addr"], wallet="wallet.dat"
)
num_blocks: int = 431
have_blocks: int = callltcrpc(0, "getblockcount")
if have_blocks < 500:
logging.info(
"Mining %d Litecoin blocks to %s", num_blocks, self.ltc_addr
)
num_blocks: int = 431
callltcrpc(
0,
"generatetoaddress",
[num_blocks - have_blocks, self.ltc_addr],
wallet="wallet.dat",
)
# 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 = callltcrpc(
0, "getnewaddress", ["mweb_addr", "mweb"], wallet="mweb"
)
callltcrpc(0, "sendtoaddress", [mweb_addr, 1.0], wallet="wallet.dat")
num_blocks = 69
have_blocks: int = callltcrpc(0, "getblockcount")
if have_blocks < 500:
logging.info(
"Mining %d Litecoin blocks to %s", num_blocks, self.ltc_addr
)
callltcrpc(
0,
"generatetoaddress",
[num_blocks - have_blocks, self.ltc_addr],
wallet="wallet.dat",
)
# 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 = callltcrpc(
0, "getnewaddress", ["mweb_addr", "mweb"], wallet="mweb"
)
callltcrpc(0, "sendtoaddress", [mweb_addr, 1.0], wallet="wallet.dat")
num_blocks = 69
have_blocks: int = callltcrpc(0, "getblockcount")
callltcrpc(
0,
"generatetoaddress",
[500 - have_blocks, self.ltc_addr],
wallet="wallet.dat",
)
if "decred" in TEST_COINS_LIST:
if RESET_TEST:
_ = calldcrrpc(0, "getnewaddress")
# assert (addr == self.dcr_addr)
self.dcr_acc = calldcrrpc(
0,
"getaccount",
[
self.dcr_addr,
],
)
calldcrrpc(
0,
"generate",
[
110,
],
)
else:
self.dcr_acc = calldcrrpc(
0,
"getaccount",
[
self.dcr_addr,
],
)
self.update_thread_dcr = threading.Thread(
target=updateThreadDCR, args=(self,)
callltcrpc(
0,
"generatetoaddress",
[500 - have_blocks, self.ltc_addr],
wallet="wallet.dat",
)
self.update_thread_dcr.start()
if "bitcoincash" in TEST_COINS_LIST:
self.bch_addr = callbchrpc(
0, "getnewaddress", ["mining_addr"], wallet="wallet.dat"
)
num_blocks: int = 200
have_blocks: int = callbchrpc(0, "getblockcount")
if have_blocks < num_blocks:
logging.info(
"Mining %d Bitcoincash blocks to %s",
num_blocks - have_blocks,
self.bch_addr,
)
callbchrpc(
0,
"generatetoaddress",
[num_blocks - have_blocks, self.bch_addr],
wallet="wallet.dat",
)
if "decred" in TEST_COINS_LIST:
if RESET_TEST:
# 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()
_ = calldcrrpc(0, "getnewaddress")
# assert (addr == self.dcr_addr)
self.dcr_acc = calldcrrpc(
0,
"getaccount",
[
self.dcr_addr,
],
)
calldcrrpc(
0,
"generate",
[
110,
],
)
else:
self.dcr_acc = calldcrrpc(
0,
"getaccount",
[
self.dcr_addr,
],
)
self.update_thread_xmr = threading.Thread(target=updateThreadXMR, args=(self,))
self.update_thread_xmr.start()
self.update_thread_dcr = threading.Thread(
target=updateThreadDCR, args=(self,)
)
self.update_thread_dcr.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, "getblockcount")
print("particl_blocks", particl_blocks)
if particl_blocks >= num_blocks:
break
self.delay_event.wait(1)
logging.info("PART blocks: %d", callpartrpc(0, "getblockcount"))
assert particl_blocks >= num_blocks
if "bitcoincash" in TEST_COINS_LIST:
self.bch_addr = callbchrpc(
0, "getnewaddress", ["mining_addr"], wallet="wallet.dat"
)
num_blocks: int = 200
have_blocks: int = callbchrpc(0, "getblockcount")
if have_blocks < num_blocks:
logging.info(
"Mining %d Bitcoincash blocks to %s",
num_blocks - have_blocks,
self.bch_addr,
)
callbchrpc(
0,
"generatetoaddress",
[num_blocks - have_blocks, self.bch_addr],
wallet="wallet.dat",
)
if "dogecoin" in TEST_COINS_LIST:
self.doge_addr = calldogerpc(
0, "getnewaddress", ["mining_addr"])
num_blocks: int = 200
have_blocks: int = calldogerpc(0, "getblockcount")
if have_blocks < num_blocks:
logging.info(
"Mining %d Dogecoin blocks to %s",
num_blocks - have_blocks,
self.bch_addr,
)
calldogerpc(
0,
"generatetoaddress",
[num_blocks - have_blocks, self.doge_addr])
if RESET_TEST:
# 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, "getblockcount")
print("particl_blocks", particl_blocks)
if particl_blocks >= num_blocks:
break
self.delay_event.wait(1)
logging.info("PART blocks: %d", callpartrpc(0, "getblockcount"))
assert particl_blocks >= num_blocks
class BaseTestWithPrepare(unittest.TestCase):
__test__ = False
update_min = int(os.getenv("UPDATE_THREAD_MIN_WAIT", "1"))
update_max = update_min * 4
xmr_update_min = int(os.getenv("XMR_UPDATE_THREAD_MIN_WAIT", "1"))
xmr_update_max = xmr_update_min * 4
dcr_update_min = int(os.getenv("DCR_UPDATE_THREAD_MIN_WAIT", "1"))
dcr_update_max = dcr_update_min * 4
delay_event = threading.Event()
update_thread = None
update_thread_xmr = None
update_thread_dcr = None
processes = []
btc_addr = None
ltc_addr = None
bch_addr = None
xmr_addr = None
dcr_addr = "SsYbXyjkKAEXXcGdFgr4u4bo4L8RkCxwQpH"
dcr_acc = None
doge_addr = None
initialised = False
@classmethod
def setUpClass(cls):
super(BaseTestWithPrepare, cls).setUpClass()
random.seed(time.time())
if os.path.exists(test_path) and not RESET_TEST:
logging.info(f"Continuing with existing directory: {test_path}")
else:
logging.info("Preparing %d nodes.", NUM_NODES)
prepare_nodes(
NUM_NODES,
TEST_COINS_LIST,
True,
{"min_sequence_lock_seconds": 60},
PORT_OFS,
)
signal.signal(
signal.SIGINT, lambda signal, frame: signal_handler(cls, signal, frame)
)
@classmethod
def tearDownClass(cls):
@ -441,12 +479,20 @@ class Test(unittest.TestCase):
cls.update_thread_dcr = None
cls.processes = []
def test_persistent(self):
self.start_processes()
def setUp(self):
if self.initialised:
return
start_processes(self)
waitForServer(self.delay_event, UI_PORT + 0)
waitForServer(self.delay_event, UI_PORT + 1)
self.initialised = True
class Test(BaseTestWithPrepare):
def test_persistent(self):
if self.run_test_persistent is False:
return
logging.info("[rm] Test::test_persistent")
while not self.delay_event.is_set():
logging.info("Looping indefinitely, ctrl+c to exit.")