Ecosystem/basicswap
1
0
Fork 0
mirror of https://github.com/basicswap/basicswap.git synced 2025-01-18 16:44:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

ui: Add wallet encryption templates.

Browse source 
tests: Test wallet encryption.
This commit is contained in:
tecnovert 2022-11-18 00:58:14 +02:00
parent 0f7df9e5f1
commit c5f31f0d1e
No known key found for this signature in database
GPG key ID: 8ED6D8750C4E3F93
28 changed files with 430 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
basicswap/basicswap.py
View file

@ -748,6 +748,9 @@ class BasicSwap(BaseApp):
if len(self.swaps_in_progress) > 0:
raise ValueError('Can\'t change passwords while swaps are in progress')
if old_password == new_password:
raise ValueError('Passwords must differ')
# Unlock all wallets to ensure they all have the same password.
for c in self.activeCoins():
ci = self.ci(c)

2
basicswap/chainparams.py
View file

@ -375,7 +375,7 @@ class CoinInterface:
return ticker
def getExchangeTicker(self, exchange_name):
return self.ticker()
return chainparams[self.coin_type()]['ticker']
def getExchangeName(self, exchange_name):
return chainparams[self.coin_type()]['name']

10
basicswap/http_server.py
View file

@ -17,6 +17,7 @@ from . import __version__
from .util import (
dumpj,
ensure,
LockedCoinError,
format_timestamp,
)
from .chainparams import (
@ -50,6 +51,7 @@ from .ui.page_offers import page_offers, page_offer, page_newoffer
from .ui.page_tor import page_tor, get_tor_established_state
from .ui.page_wallet import page_wallets, page_wallet
from .ui.page_settings import page_settings
from .ui.page_encryption import page_changepassword, page_unlock, page_lock
env = Environment(loader=PackageLoader('basicswap', 'templates'))
env.filters['formatts'] = format_timestamp
@ -623,9 +625,17 @@ class HttpHandler(BaseHTTPRequestHandler):
return page_automation_strategy_new(self, url_split, post_string)
if page == 'shutdown':
return self.page_shutdown(url_split, post_string)
if page == 'changepassword':
return page_changepassword(self, url_split, post_string)
if page == 'unlock':
return page_unlock(self, url_split, post_string)
if page == 'lock':
return page_lock(self, url_split, post_string)
if page != '':
return self.page_404(url_split)
return self.page_index(url_split)
except LockedCoinError:
return page_unlock(self, url_split, post_string)
except Exception as ex:
if swap_client.debug is True:
swap_client.log.error(traceback.format_exc())

2
basicswap/interface/btc.py
View file

@ -1305,6 +1305,8 @@ class BTCInterface(CoinInterface):
def changeWalletPassword(self, old_password, new_password):
self._log.info('changeWalletPassword - {}'.format(self.ticker()))
if old_password == '':
if self.isWalletEncrypted():
raise ValueError('Old password must be set')
return self.rpc_callback('encryptwallet', [new_password])
self.rpc_callback('walletpassphrasechange', [old_password, new_password])

2
basicswap/js_server.py
View file

@ -481,7 +481,7 @@ def js_unlock(self, url_split, post_string, is_json):
def js_lock(self, url_split, post_string, is_json):
swap_client = self.server.swap_client
post_data = getFormData(post_string, is_json)
post_data = {} if post_string == '' else getFormData(post_string, is_json)
if have_data_entry(post_data, 'coin'):
coin = getCoinType(get_data_entry(post_data, 'coin'))

27
basicswap/templates/changepassword.html Normal file
View file

@ -0,0 +1,27 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel=icon sizes="32x32" type="image/png" href="/static/images/favicon-32.png">
<title>{{ title }}</title>
</head>
<body>
<h2>Change password</h2>
{% for m in messages %}
<p>{{ m }}</p>
{% endfor %}
{% for m in err_messages %}
<p class="error_msg">Error: {{ m }}</p>
{% endfor %}
<form method="post">
<table>
<tr><td>Old Password</td><td><input type="text" name="oldpassword"></td></tr>
<tr><td>New Password</td><td><input type="text" name="newpassword"></td></tr>
<tr><td>Confirm Password</td><td><input type="text" name="confirmpassword"></td></tr>
<tr><td><input type="submit" name="unlock" value="Unlock"></td></tr>
</table>
<input type="hidden" name="formid" value="{{ form_id }}">
</form>
<p><a href="/">home</a></p>
</body>
</html>

23
basicswap/templates/unlock.html Normal file
View file

@ -0,0 +1,23 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel=icon sizes="32x32" type="image/png" href="/static/images/favicon-32.png">
<title>{{ title }}</title>
</head>
<body>
<h2>Unlock wallet</h2>
{% for m in messages %}
<p>{{ m }}</p>
{% endfor %}
{% for m in err_messages %}
<p class="error_msg">Error: {{ m }}</p>
{% endfor %}
<form method="post">
<input type="text" name="password"><br/>
<input type="submit" name="unlock" value="Unlock">
<input type="hidden" name="formid" value="{{ form_id }}">
</form>
<p><a href="/">home</a></p>
</body>
</html>

87
basicswap/ui/page_encryption.py Normal file
View file

@ -0,0 +1,87 @@
# -*- 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.
from .util import (
get_data_entry_or,
)
def page_changepassword(self, url_split, post_string):
server = self.server
swap_client = server.swap_client
swap_client.checkSystemStatus()
messages = []
err_messages = []
form_data = self.checkForm(post_string, 'changepassword', err_messages)
if form_data:
old_password = get_data_entry_or(form_data, 'oldpassword', '')
new_password = get_data_entry_or(form_data, 'newpassword', '')
confirm_password = get_data_entry_or(form_data, 'confirmpassword', '')
try:
if new_password == '':
raise ValueError('New password must be entered.')
if new_password != confirm_password:
raise ValueError('New password and confirm password must match.')
swap_client.changeWalletPasswords(old_password, new_password)
messages.append('Password changed')
except Exception as e:
err_messages.append(str(e))
template = server.env.get_template('changepassword.html')
return self.render_template(template, {
'messages': messages,
'err_messages': err_messages,
})
def page_unlock(self, url_split, post_string):
server = self.server
swap_client = server.swap_client
messages = []
err_messages = []
form_data = self.checkForm(post_string, 'unlock', err_messages)
if form_data:
password = get_data_entry_or(form_data, 'password', '')
try:
if password == '':
raise ValueError('Password must be entered.')
swap_client.unlockWallets(password)
self.send_response(302)
self.send_header('Location', '/')
self.end_headers()
return bytes()
except Exception as e:
err_messages.append(str(e))
template = server.env.get_template('unlock.html')
return self.render_template(template, {
'messages': messages,
'err_messages': err_messages,
})
def page_lock(self, url_split, post_string):
server = self.server
swap_client = server.swap_client
swap_client.checkSystemStatus()
swap_client.lockWallets()
messages = []
err_messages = []
template = server.env.get_template('info.html')
return self.render_template(template, {
'messages': messages,
'err_messages': err_messages,
'message_str': 'Wallets locked'
})

51
tests/basicswap/common.py
View file

@ -7,11 +7,11 @@
import os
import json
import urllib
import signal
import logging
from urllib.request import urlopen
from .util import read_json_api
from basicswap.rpc import callrpc
from basicswap.contrib.rpcauth import generate_salt, password_to_hmac
from bin.basicswap_prepare import downloadPIVXParams
@ -215,38 +215,6 @@ def wait_for_in_progress(delay_event, swap_client, bid_id, sent=False):
raise ValueError('wait_for_in_progress timed out.')
def post_json_req(url, json_data):
req = urllib.request.Request(url)
req.add_header('Content-Type', 'application/json; charset=utf-8')
post_bytes = json.dumps(json_data).encode('utf-8')
req.add_header('Content-Length', len(post_bytes))
return urlopen(req, post_bytes, timeout=300).read()
def read_text_api(port, path=None):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
return urlopen(url, timeout=300).read().decode('utf-8')
def read_json_api(port, path=None, json_data=None):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
if json_data is not None:
return json.loads(post_json_req(url, json_data))
return json.loads(urlopen(url, timeout=300).read())
def post_json_api(port, path, json_data):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
return json.loads(post_json_req(url, json_data))
def wait_for_none_active(delay_event, port, wait_for=30):
for i in range(wait_for):
if delay_event.is_set():
@ -258,19 +226,6 @@ def wait_for_none_active(delay_event, port, wait_for=30):
raise ValueError('wait_for_none_active timed out.')
def waitForServer(delay_event, port, wait_for=20):
for i in range(wait_for):
if delay_event.is_set():
raise ValueError('Test stopped.')
try:
delay_event.wait(1)
summary = read_json_api(port)
return
except Exception as e:
print('waitForServer, error:', str(e))
raise ValueError('waitForServer failed')
def waitForNumOffers(delay_event, port, offers, wait_for=20):
for i in range(wait_for):
if delay_event.is_set():
@ -321,10 +276,6 @@ 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)

4
tests/basicswap/common_xmr.py
View file

@ -22,8 +22,10 @@ from basicswap.rpc_xmr import (
callrpc_xmr_na,
)
from tests.basicswap.mnemonics import mnemonics
from tests.basicswap.common import (
from tests.basicswap.util import (
waitForServer,
)
from tests.basicswap.common import (
BASE_PORT, BASE_RPC_PORT,
BTC_BASE_PORT, BTC_BASE_RPC_PORT, BTC_BASE_TOR_PORT,
LTC_BASE_PORT,

4
tests/basicswap/extended/test_dash.py
View file

@ -48,6 +48,9 @@ from basicswap.contrib.key import (
from basicswap.http_server import (
HttpThread,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
checkForks,
stopDaemons,
@ -55,7 +58,6 @@ from tests.basicswap.common import (
wait_for_bid,
wait_for_bid_tx_state,
wait_for_in_progress,
read_json_api,
TEST_HTTP_HOST,
TEST_HTTP_PORT,
BASE_PORT,

4
tests/basicswap/extended/test_firo.py
View file

@ -29,11 +29,13 @@ from basicswap.rpc import (
callrpc_cli,
waitForRPC,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
stopDaemons,
wait_for_bid,
make_rpc_func,
read_json_api,
TEST_HTTP_PORT,
wait_for_offer,
wait_for_in_progress,

4
tests/basicswap/extended/test_network.py
View file

@ -39,13 +39,15 @@ from basicswap.contrib.key import (
from basicswap.http_server import (
HttpThread,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
prepareDataDir,
make_rpc_func,
checkForks,
stopDaemons,
delay_for,
read_json_api,
TEST_HTTP_HOST,
TEST_HTTP_PORT,
BASE_P2P_PORT,

4
tests/basicswap/extended/test_nmc.py
View file

@ -47,6 +47,9 @@ from basicswap.contrib.key import (
from basicswap.http_server import (
HttpThread,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
checkForks,
stopDaemons,
@ -54,7 +57,6 @@ from tests.basicswap.common import (
wait_for_bid,
wait_for_bid_tx_state,
wait_for_in_progress,
read_json_api,
TEST_HTTP_HOST,
TEST_HTTP_PORT,
BASE_PORT,

4
tests/basicswap/extended/test_pivx.py
View file

@ -47,6 +47,9 @@ from basicswap.contrib.key import (
from basicswap.http_server import (
HttpThread,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
checkForks,
stopDaemons,
@ -54,7 +57,6 @@ from tests.basicswap.common import (
wait_for_bid,
wait_for_bid_tx_state,
wait_for_in_progress,
read_json_api,
TEST_HTTP_HOST,
TEST_HTTP_PORT,
BASE_PORT,

201
tests/basicswap/extended/test_prepare.py
View file

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2019 tecnovert
# Copyright (c) 2019-2022 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
@ -11,71 +11,192 @@ import json
import shutil
import logging
import unittest
import threading
import multiprocessing
from io import StringIO
from unittest.mock import patch
import basicswap.config as cfg
import bin.basicswap_prepare as prepareSystem
test_path = os.path.expanduser(os.getenv('TEST_PREPARE_PATH', '~/test_basicswap'))
from tests.basicswap.util import (
read_json_api,
waitForServer,
)
bin_path = os.path.expanduser(os.getenv('TEST_BIN_PATH', ''))
test_base_path = os.path.expanduser(os.getenv('TEST_PREPARE_PATH', '~/test_basicswap'))
test_path_plain = os.path.join(test_base_path, 'plain')
test_path_encrypted = os.path.join(test_base_path, 'encrypted')
test_path_encrypt = os.path.join(test_base_path, 'encrypt')
delay_event = threading.Event()
logger = logging.getLogger()
logger.level = logging.DEBUG
if not len(logger.handlers):
logger.addHandler(logging.StreamHandler(sys.stdout))
logger.addHandler(logging.StreamHandler(sys.stdout))
def start_prepare(args, env_pairs=[]):
for pair in env_pairs:
os.environ[pair[0]] = pair[1]
print(pair[0], os.environ[pair[0]])
import bin.basicswap_prepare as prepareSystemThread
with patch.object(sys, 'argv', args):
prepareSystemThread.main()
del prepareSystemThread
def start_run(args, env_pairs=[]):
for pair in env_pairs:
os.environ[pair[0]] = pair[1]
print(pair[0], os.environ[pair[0]])
import bin.basicswap_run as runSystemThread
with patch.object(sys, 'argv', args):
runSystemThread.main()
del runSystemThread
class Test(unittest.TestCase):
@classmethod
def tearDownClass(self):
try:
shutil.rmtree(test_path)
for test_dir in (test_path_plain, test_path_encrypted, test_path_encrypt):
if os.path.exists(test_dir):
shutil.rmtree(test_dir)
except Exception as ex:
logger.warning('tearDownClass %s', str(ex))
super(Test, self).tearDownClass()
def test(self):
testargs = ['basicswap-prepare', '-datadir=' + test_path, '-withcoin=litecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
def test_plain(self):
if os.path.exists(test_path_plain):
shutil.rmtree(test_path_plain)
if bin_path != '':
os.makedirs(test_path_plain)
os.symlink(bin_path, os.path.join(test_path_plain, 'bin'))
config_path = os.path.join(test_path, cfg.CONFIG_FILENAME)
testargs = ('basicswap-prepare', '-datadir=' + test_path_plain, '-withcoin=litecoin')
process = multiprocessing.Process(target=start_prepare, args=(testargs,))
process.start()
process.join()
config_path = os.path.join(test_path_plain, cfg.CONFIG_FILENAME)
self.assertTrue(os.path.exists(config_path))
logger.info('Test no overwrite')
testargs = ['basicswap-prepare', '-datadir=' + test_path, '-withcoin=litecoin']
with patch('sys.stderr', new=StringIO()) as fake_stderr:
import bin.basicswap_prepare as prepareSystem
try:
logging.info('Test no overwrite')
testargs = ['basicswap-prepare', '-datadir=' + test_path_plain, '-withcoin=litecoin']
with patch('sys.stderr', new=StringIO()) as fake_stderr:
with patch.object(sys, 'argv', testargs):
with self.assertRaises(SystemExit) as cm:
prepareSystem.main()
self.assertEqual(cm.exception.code, 1)
logger.info('fake_stderr.getvalue() %s', fake_stderr.getvalue())
self.assertTrue('exists, exiting' in fake_stderr.getvalue())
logger.info('Test addcoin new')
testargs = ['basicswap-prepare', '-datadir=' + test_path_plain, '-addcoin=namecoin']
with patch.object(sys, 'argv', testargs):
with self.assertRaises(SystemExit) as cm:
prepareSystem.main()
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'rpc')
self.assertEqual(cm.exception.code, 1)
logger.info('fake_stderr.getvalue() %s', fake_stderr.getvalue())
self.assertTrue('exists, exiting' in fake_stderr.getvalue())
logger.info('Test disablecoin')
testargs = ['basicswap-prepare', '-datadir=' + test_path_plain, '-disablecoin=namecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'none')
logger.info('Test addcoin new')
testargs = ['basicswap-prepare', '-datadir=' + test_path, '-addcoin=namecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'rpc')
logger.info('Test addcoin existing')
testargs = ['basicswap-prepare', '-datadir=' + test_path_plain, '-addcoin=namecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'rpc')
finally:
del prepareSystem
logger.info('Test disablecoin')
testargs = ['basicswap-prepare', '-datadir=' + test_path, '-disablecoin=namecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'none')
def test_encrypted(self):
if os.path.exists(test_path_encrypted):
shutil.rmtree(test_path_encrypted)
if bin_path != '':
os.makedirs(test_path_encrypted)
os.symlink(bin_path, os.path.join(test_path_encrypted, 'bin'))
logger.info('Test addcoin existing')
testargs = ['basicswap-prepare', '-datadir=' + test_path, '-addcoin=namecoin']
with patch.object(sys, 'argv', testargs):
prepareSystem.main()
with open(config_path) as fs:
settings = json.load(fs)
self.assertTrue(settings['chainclients']['namecoin']['connection_type'] == 'rpc')
env_vars = [('WALLET_ENCRYPTION_PWD', 'test123'), ]
testargs = ('basicswap-prepare', '-datadir=' + test_path_encrypted, '-withcoin=litecoin,monero')
process = multiprocessing.Process(target=start_prepare, args=(testargs, env_vars))
process.start()
process.join()
assert (process.exitcode == 0)
logger.info('Should not be able to add a coin without setting WALLET_ENCRYPTION_PWD')
testargs = ('basicswap-prepare', '-datadir=' + test_path_encrypted, '-addcoin=bitcoin')
process = multiprocessing.Process(target=start_prepare, args=(testargs, []))
process.start()
process.join()
assert (process.exitcode == 1)
testargs = ('basicswap-prepare', '-datadir=' + test_path_encrypted, '-addcoin=bitcoin')
process = multiprocessing.Process(target=start_prepare, args=(testargs, env_vars))
process.start()
process.join()
assert (process.exitcode == 0)
def test_encrypt(self):
if os.path.exists(test_path_encrypt):
shutil.rmtree(test_path_encrypt)
if bin_path != '':
os.makedirs(test_path_encrypt)
os.symlink(bin_path, os.path.join(test_path_encrypt, 'bin'))
testargs = ('basicswap-prepare', '-regtest=1', '-datadir=' + test_path_encrypt, '-withcoin=litecoin,monero')
process = multiprocessing.Process(target=start_prepare, args=(testargs, ))
process.start()
process.join()
assert (process.exitcode == 0)
logger.info('basicswap-run should fail if WALLET_ENCRYPTION_PWD is set')
env_vars = [('WALLET_ENCRYPTION_PWD', 'test123'), ]
testargs = ('basicswap-run', '-regtest=1', '-datadir=' + test_path_encrypt)
process = multiprocessing.Process(target=start_run, args=(testargs, env_vars))
process.start()
process.join()
assert (process.exitcode == 1)
testargs = ('basicswap-run', '-regtest=1', '-datadir=' + test_path_encrypt)
process = multiprocessing.Process(target=start_run, args=(testargs, ))
process.start()
waitForServer(delay_event, 12700)
rv = read_json_api(12700, 'setpassword', {'oldpassword': 'wrongpass', 'newpassword': 'test123'})
assert ('error' in rv)
rv = read_json_api(12700, 'setpassword', {'oldpassword': '', 'newpassword': 'test123'})
assert ('success' in rv)
rv = read_json_api(12700, 'setpassword', {'oldpassword': 'test123', 'newpassword': 'next123'})
assert ('success' in rv)
rv = read_json_api(12700, 'lock')
assert ('success' in rv)
rv = read_json_api(12700, 'wallets')
assert ('error' in rv)
rv = read_json_api(12700, 'unlock', {'password': 'next123'})
assert ('success' in rv)
rv = read_json_api(12700, 'wallets')
assert ('PART' in rv)
process.terminate()
process.join()
assert (process.exitcode == 0)
if __name__ == '__main__':

2
tests/basicswap/extended/test_wallet_init.py
View file

@ -27,7 +27,7 @@ import multiprocessing
from unittest.mock import patch
from tests.basicswap.mnemonics import mnemonics
from tests.basicswap.common import (
from tests.basicswap.util import (
read_json_api,
waitForServer,
)

4
tests/basicswap/extended/test_wallet_restore.py
View file

@ -26,10 +26,12 @@ import traceback
import multiprocessing
from unittest.mock import patch
from tests.basicswap.common import (
from tests.basicswap.util import (
read_json_api,
post_json_api,
waitForServer,
)
from tests.basicswap.common import (
waitForNumOffers,
waitForNumBids,
)

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

@ -34,11 +34,13 @@ from basicswap.rpc import (
callrpc,
)
from tests.basicswap.common import (
BASE_RPC_PORT,
BTC_BASE_RPC_PORT,
)
from tests.basicswap.util import (
make_boolean,
read_json_api,
waitForServer,
BASE_RPC_PORT,
BTC_BASE_RPC_PORT,
)
from tests.basicswap.common_xmr import (
prepare_nodes,

4
tests/basicswap/test_btc_xmr.py
View file

@ -22,9 +22,11 @@ from basicswap.util import (
make_int,
format_amount,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
wait_for_bid,
read_json_api,
wait_for_offer,
wait_for_none_active,
BTC_BASE_RPC_PORT,

4
tests/basicswap/test_ltc_xmr.py
View file

@ -17,9 +17,11 @@ from basicswap.basicswap import (
from basicswap.util import (
COIN,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
wait_for_bid,
read_json_api,
wait_for_offer,
wait_for_in_progress,
LTC_BASE_RPC_PORT,

6
tests/basicswap/test_partblind_xmr.py
View file

@ -25,13 +25,15 @@ from basicswap.util import (
make_int,
format_amount,
)
from tests.basicswap.util import (
post_json_req,
read_json_api,
)
from tests.basicswap.common import (
wait_for_bid,
read_json_api,
wait_for_offer,
wait_for_none_active,
wait_for_balance,
post_json_req,
)
from .test_xmr import BaseTest, test_delay_event

4
tests/basicswap/test_reload.py
View file

@ -26,10 +26,12 @@ from unittest.mock import patch
from basicswap.rpc import (
callrpc_cli,
)
from tests.basicswap.common import (
from tests.basicswap.util import (
read_json_api,
post_json_api,
waitForServer,
)
from tests.basicswap.common import (
waitForNumOffers,
waitForNumBids,
waitForNumSwapping,

4
tests/basicswap/test_run.py
View file

@ -36,13 +36,15 @@ from basicswap.util import (
make_int,
format_amount,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
wait_for_offer,
wait_for_bid,
wait_for_balance,
wait_for_bid_tx_state,
wait_for_in_progress,
read_json_api,
TEST_HTTP_PORT,
LTC_BASE_RPC_PORT,
BTC_BASE_RPC_PORT,

10
tests/basicswap/test_xmr.py
View file

@ -57,9 +57,15 @@ from basicswap.contrib.key import (
from basicswap.http_server import (
HttpThread,
)
from tests.basicswap.util import (
make_boolean,
post_json_req,
)
from tests.basicswap.util import (
read_json_api,
)
from tests.basicswap.common import (
prepareDataDir,
make_boolean,
make_rpc_func,
checkForks,
stopDaemons,
@ -69,8 +75,6 @@ from tests.basicswap.common import (
wait_for_no_offer,
wait_for_none_active,
wait_for_balance,
post_json_req,
read_json_api,
compare_bid_states,
extract_states_from_xu_file,
TEST_HTTP_HOST,

4
tests/basicswap/test_xmr_bids_offline.py
View file

@ -23,9 +23,11 @@ import multiprocessing
from urllib import parse
from urllib.request import urlopen
from tests.basicswap.common import (
from tests.basicswap.util import (
read_json_api,
waitForServer,
)
from tests.basicswap.common import (
waitForNumOffers,
waitForNumBids,
)

4
tests/basicswap/test_xmr_reload.py
View file

@ -20,10 +20,12 @@ import logging
import unittest
import multiprocessing
from tests.basicswap.common import (
from tests.basicswap.util import (
read_json_api,
post_json_api,
waitForServer,
)
from tests.basicswap.common import (
waitForNumOffers,
waitForNumBids,
waitForNumSwapping,

59
tests/basicswap/util.py Normal file
View file

@ -0,0 +1,59 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2022 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php.
import json
import urllib
from urllib.request import urlopen
def make_boolean(s):
return s.lower() in ['1', 'true']
def post_json_req(url, json_data):
req = urllib.request.Request(url)
req.add_header('Content-Type', 'application/json; charset=utf-8')
post_bytes = json.dumps(json_data).encode('utf-8')
req.add_header('Content-Length', len(post_bytes))
return urlopen(req, post_bytes, timeout=300).read()
def read_text_api(port, path=None):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
return urlopen(url, timeout=300).read().decode('utf-8')
def read_json_api(port, path=None, json_data=None):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
if json_data is not None:
return json.loads(post_json_req(url, json_data))
return json.loads(urlopen(url, timeout=300).read())
def post_json_api(port, path, json_data):
url = f'http://127.0.0.1:{port}/json'
if path is not None:
url += '/' + path
return json.loads(post_json_req(url, json_data))
def waitForServer(delay_event, port, wait_for=20):
for i in range(wait_for):
if delay_event.is_set():
raise ValueError('Test stopped.')
try:
delay_event.wait(1)
summary = read_json_api(port)
return
except Exception as e:
print('waitForServer, error:', str(e))
raise ValueError('waitForServer failed')