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

PART -> XMR tests pass.

Browse Source
This commit is contained in:
tecnovert
2020-11-27 19:52:26 +02:00
parent 0e2011e085
commit bc60527940
11 changed files with 566 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
tests/basicswap/test_other.py
View File

@@ -6,6 +6,7 @@
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import unittest
import secrets
import basicswap.contrib.ed25519_fast as edf
import basicswap.ed25519_fast_util as edu
@@ -13,6 +14,14 @@ import basicswap.ed25519_fast_util as edu
from basicswap.ecc_util import i2b
from coincurve.ed25519 import ed25519_get_pubkey
from coincurve.ecdsaotves import (
ecdsaotves_enc_sign,
ecdsaotves_enc_verify,
ecdsaotves_dec_sig,
ecdsaotves_rec_enc_key)
from basicswap.interface_btc import BTCInterface
from basicswap.interface_xmr import XMRInterface
from basicswap.util import (
SerialiseNum,
@@ -30,6 +39,9 @@ from basicswap.basicswap import (
SEQUENCE_LOCK_TIME,
)
from basicswap.ecc_util import (
i2b)
class Test(unittest.TestCase):
def test_serialise_num(self):
@@ -152,6 +164,36 @@ class Test(unittest.TestCase):
pubkey_test = ed25519_get_pubkey(privkey_bytes)
assert(pubkey == pubkey_test)
def test_ecdsa_otves(self):
coin_settings = {'rpcport': 0, 'rpcauth': 'none', 'blocks_confirmed': 1}
ci = BTCInterface(coin_settings, 'regtest')
vk_sign = i2b(ci.getNewSecretKey())
vk_encrypt = i2b(ci.getNewSecretKey())
pk_sign = ci.getPubkey(vk_sign)
pk_encrypt = ci.getPubkey(vk_encrypt)
sign_hash = secrets.token_bytes(32)
cipher_text = ecdsaotves_enc_sign(vk_sign, pk_encrypt, sign_hash)
assert(ecdsaotves_enc_verify(pk_sign, pk_encrypt, sign_hash, cipher_text))
sig = ecdsaotves_dec_sig(vk_encrypt, cipher_text)
assert(ci.verifySig(pk_sign, sign_hash, sig))
recovered_key = ecdsaotves_rec_enc_key(pk_encrypt, cipher_text, sig)
assert(vk_encrypt == recovered_key)
def test_dleag(self):
coin_settings = {'rpcport': 0, 'walletrpcport': 0, 'walletrpcauth': 'none', 'blocks_confirmed': 1}
ci = XMRInterface(coin_settings, 'regtest')
key = i2b(ci.getNewSecretKey())
proof = ci.proveDLEAG(key)
assert(ci.verifyDLEAG(proof))
if __name__ == '__main__':
unittest.main()

115
tests/basicswap/test_xmr.py
View File

@@ -6,7 +6,6 @@
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import os
import sys
import json
import time
import shutil
@@ -17,14 +16,6 @@ import traceback
import threading
import subprocess
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 (
@@ -33,6 +24,7 @@ from basicswap.basicswap import (
SwapTypes,
BidStates,
TxStates,
DebugTypes,
SEQUENCE_LOCK_BLOCKS,
)
from basicswap.util import (
@@ -201,7 +193,7 @@ def prepare_swapclient_dir(datadir, node_id, network_key, network_pubkey):
'zmqhost': 'tcp://127.0.0.1',
'zmqport': BASE_ZMQ_PORT + node_id,
'htmlhost': 'localhost',
'htmlport': 12700 + node_id,
'htmlport': TEST_HTTP_PORT + node_id,
'network_key': network_key,
'network_pubkey': network_pubkey,
'chainclients': {
@@ -303,6 +295,7 @@ def callnoderpc(node_id, method, params=[], wallet=None, base_rpc_port=BASE_RPC_
auth = 'test{0}:test_pass{0}'.format(node_id)
return callrpc(base_rpc_port + node_id, auth, method, params, wallet)
def run_coins_loop(cls):
global stop_test
while not stop_test:
@@ -313,6 +306,7 @@ def run_coins_loop(cls):
callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': cls.xmr_addr, 'amount_of_blocks': 1})
time.sleep(1.0)
def run_loop(cls):
global stop_test
while not stop_test:
@@ -547,8 +541,6 @@ class Test(unittest.TestCase):
offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 100 * COIN, 10 * XMR_COIN, 100 * COIN, SwapTypes.XMR_SWAP)
self.wait_for_offer(swap_clients[1], offer_id)
offer = swap_clients[1].getOffer(offer_id)
offers = swap_clients[1].listOffers(filters={'offer_id': offer_id})
assert(len(offers) == 1)
offer = offers[0]
@@ -572,7 +564,106 @@ class Test(unittest.TestCase):
self.wait_for_bid(swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=180)
self.wait_for_bid(swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True)
def test_02_leader_recover_a_lock_tx(self):
logging.info('---------- Test PART to XMR leader recovers coin a lock tx')
swap_clients = self.swap_clients
js_w0_before = json.loads(urlopen('http://localhost:1800/json/wallets').read())
offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 101 * COIN, 1.1 * XMR_COIN, 101 * COIN, SwapTypes.XMR_SWAP,
lock_type=SEQUENCE_LOCK_BLOCKS, lock_value=12)
self.wait_for_offer(swap_clients[1], offer_id)
offer = swap_clients[1].getOffer(offer_id)
bid_id = swap_clients[1].postXmrBid(offer_id, offer.amount_from)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.BID_RECEIVED)
bid, xmr_swap = swap_clients[0].getXmrBid(bid_id)
assert(xmr_swap)
swap_clients[1].setBidDebugInd(bid_id, DebugTypes.BID_STOP_AFTER_COIN_A_LOCK)
swap_clients[0].acceptXmrBid(bid_id)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.XMR_SWAP_FAILED_REFUNDED, wait_for=180)
self.wait_for_bid(swap_clients[1], bid_id, BidStates.XMR_SWAP_FAILED_REFUNDED, sent=True)
js_w0_after = json.loads(urlopen('http://localhost:1800/json/wallets').read())
print('[rm] js_w0_before', json.dumps(js_w0_before))
print('[rm] js_w0_after', json.dumps(js_w0_after))
def test_03_follower_recover_a_lock_tx(self):
logging.info('---------- Test PART to XMR follower recovers coin a lock tx')
swap_clients = self.swap_clients
js_w0_before = json.loads(urlopen('http://localhost:1800/json/wallets').read())
offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 101 * COIN, 1.1 * XMR_COIN, 101 * COIN, SwapTypes.XMR_SWAP,
lock_type=SEQUENCE_LOCK_BLOCKS, lock_value=12)
self.wait_for_offer(swap_clients[1], offer_id)
offer = swap_clients[1].getOffer(offer_id)
bid_id = swap_clients[1].postXmrBid(offer_id, offer.amount_from)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.BID_RECEIVED)
bid, xmr_swap = swap_clients[0].getXmrBid(bid_id)
assert(xmr_swap)
swap_clients[1].setBidDebugInd(bid_id, DebugTypes.BID_STOP_AFTER_COIN_A_LOCK)
swap_clients[0].setBidDebugInd(bid_id, DebugTypes.BID_DONT_SPEND_COIN_A_LOCK_REFUND)
swap_clients[0].acceptXmrBid(bid_id)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.BID_ABANDONED, wait_for=180)
self.wait_for_bid(swap_clients[1], bid_id, BidStates.XMR_SWAP_FAILED_SWIPED, sent=True)
js_w0_after = json.loads(urlopen('http://localhost:1800/json/wallets').read())
def test_04_follower_recover_b_lock_tx(self):
logging.info('---------- Test PART to XMR follower recovers coin b lock tx')
swap_clients = self.swap_clients
offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 101 * COIN, 1.1 * XMR_COIN, 101 * COIN, SwapTypes.XMR_SWAP,
lock_type=SEQUENCE_LOCK_BLOCKS, lock_value=18)
self.wait_for_offer(swap_clients[1], offer_id)
offer = swap_clients[1].getOffer(offer_id)
bid_id = swap_clients[1].postXmrBid(offer_id, offer.amount_from)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.BID_RECEIVED)
bid, xmr_swap = swap_clients[0].getXmrBid(bid_id)
assert(xmr_swap)
swap_clients[1].setBidDebugInd(bid_id, DebugTypes.CREATE_INVALID_COIN_B_LOCK)
swap_clients[0].acceptXmrBid(bid_id)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.XMR_SWAP_FAILED_REFUNDED, wait_for=180)
self.wait_for_bid(swap_clients[1], bid_id, BidStates.XMR_SWAP_FAILED_REFUNDED, sent=True)
def test_05_btc_xmr(self):
logging.info('---------- Test BTC to XMR')
swap_clients = self.swap_clients
offer_id = swap_clients[0].postOffer(Coins.BTC, Coins.XMR, 10 * COIN, 100 * XMR_COIN, 100 * COIN, SwapTypes.XMR_SWAP)
self.wait_for_offer(swap_clients[1], offer_id)
offers = swap_clients[1].listOffers(filters={'offer_id': offer_id})
offer = offers[0]
bid_id = swap_clients[1].postXmrBid(offer_id, offer.amount_from)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.BID_RECEIVED)
bid, xmr_swap = swap_clients[0].getXmrBid(bid_id)
assert(xmr_swap)
swap_clients[0].acceptXmrBid(bid_id)
self.wait_for_bid(swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=180)
self.wait_for_bid(swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True)
if __name__ == '__main__':