DragonX/dragonx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Prolonged Proxy timeout

Browse Source 
Added support for multiple test chains
This commit is contained in:
SirSevenG
2019-08-02 03:05:47 +07:00
parent 07bb419492
commit 7a4f8b3401
2 changed files with 73 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
qa/rpc-tests/nspv_client_test.py Normal file → Executable file
View File

@@ -7,19 +7,48 @@ import time
''' '''
specify chain ticker (daemon should be up), wif which will be imported and address to which you want to broadcast specify chain ticker (daemon should be up), wif which will be imported and address to which you want to broadcast
change chain parameters if needed or add a new chain to test below
added 1 second sleep after each case to surely not face the nSPV server limitation (1 call/second) added 1 second sleep after each case to surely not face the nSPV server limitation (1 call/second)
''' '''
wif = '' wif = ''
dest_address = '' dest_address = ''
amount = '0.01' amount = '0.01'
chain = 'ILN' chain = 'KMD'
if not wif or not dest_address: if not wif or not dest_address:
raise Exception("Please set test wif and address to send transactions to") raise Exception("Please set test wif and address to send transactions to")
rpc_proxy = tuilib.def_credentials(chain) rpc_proxy = tuilib.def_credentials(chain)
chain_params = {"KMD": {
'tx_list_address': 'RGShWG446Pv24CKzzxjA23obrzYwNbs1kA',
'min_chain_height': 1468080,
'notarization_height': '1468000',
'prev_notarization_h': 1467980,
'next_notarization_h': 1468020,
'hdrs_proof_low': '1468100',
'hdrs_proof_high': '1468200',
'numhdrs_expected': 151,
'tx_proof_id': 'f7beb36a65bc5bcbc9c8f398345aab7948160493955eb4a1f05da08c4ac3784f',
'tx_spent_height': 1456212,
'tx_proof_height': '1468520',
},
"ILN": {
'tx_list_address': 'RUp3xudmdTtxvaRnt3oq78FJBjotXy55uu',
'min_chain_height': 3689,
'notarization_height': '2000',
'prev_notarization_h': 1998,
'next_notarization_h': 2008,
'hdrs_proof_low': '2000',
'hdrs_proof_high': '2100',
'numhdrs_expected': 113,
'tx_proof_id': '67ffe0eaecd6081de04675c492a59090b573ee78955c4e8a85b8ac0be0e8e418',
'tx_spent_height': 2681,
'tx_proof_height': '2690',
}
}
class TestNspvClient(unittest.TestCase): class TestNspvClient(unittest.TestCase):
@@ -37,31 +66,32 @@ class TestNspvClient(unittest.TestCase):
result = rpc_proxy.nspv_listtransactions() result = rpc_proxy.nspv_listtransactions()
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
time.sleep(1) time.sleep(1)
result = rpc_proxy.nspv_listtransactions("RUp3xudmdTtxvaRnt3oq78FJBjotXy55uu") result = rpc_proxy.nspv_listtransactions(chain_params.get(chain).get("tx_list_address"))
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertEqual(result["address"], "RUp3xudmdTtxvaRnt3oq78FJBjotXy55uu") self.assertEqual(result["address"], chain_params.get(chain).get("tx_list_address"))
rpc_proxy.nspv_logout() rpc_proxy.nspv_logout()
def test_nspv_getinfo(self): def test_nspv_getinfo(self):
print("testing nspv_getinfo") print("testing nspv_getinfo")
result = rpc_proxy.nspv_getinfo() result = rpc_proxy.nspv_getinfo()
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertGreater(result["height"], 2689) self.assertGreater(result["height"], chain_params.get(chain).get("min_chain_height"))
time.sleep(1) time.sleep(1)
def test_nspv_notarizations(self): def test_nspv_notarizations(self):
print("testing nspv_notarizations") print("testing nspv_notarizations")
result = rpc_proxy.nspv_notarizations("2000") result = rpc_proxy.nspv_notarizations(chain_params.get(chain).get("notarization_height"))
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertEqual(result["prev"]["notarized_height"], 1998) self.assertEqual(result["prev"]["notarized_height"], chain_params.get(chain).get("prev_notarization_h"))
self.assertEqual(result["next"]["notarized_height"], 2008) self.assertEqual(result["next"]["notarized_height"], chain_params.get(chain).get("next_notarization_h"))
time.sleep(1) time.sleep(1)
def test_nspv_hdrsproof(self): def test_nspv_hdrsproof(self):
print("testing nspv_hdrsproof") print("testing nspv_hdrsproof")
result = rpc_proxy.nspv_hdrsproof("2000", "2100") result = rpc_proxy.nspv_hdrsproof(chain_params.get(chain).get("hdrs_proof_low"),
chain_params.get(chain).get("hdrs_proof_high"))
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertEqual(result["numhdrs"], 113) self.assertEqual(result["numhdrs"], chain_params.get(chain).get("numhdrs_expected"))
time.sleep(1) time.sleep(1)
def test_nspv_login(self): def test_nspv_login(self):
@@ -76,9 +106,9 @@ class TestNspvClient(unittest.TestCase):
result = rpc_proxy.nspv_listunspent() result = rpc_proxy.nspv_listunspent()
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
time.sleep(1) time.sleep(1)
result = rpc_proxy.nspv_listunspent("RUp3xudmdTtxvaRnt3oq78FJBjotXy55uu") result = rpc_proxy.nspv_listunspent(chain_params.get(chain).get("tx_list_address"))
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertEqual(result["address"], "RUp3xudmdTtxvaRnt3oq78FJBjotXy55uu") self.assertEqual(result["address"], chain_params.get(chain).get("tx_list_address"))
def test_nspv_spend(self): def test_nspv_spend(self):
print("testing nspv_spend") print("testing nspv_spend")
@@ -113,25 +143,26 @@ class TestNspvClient(unittest.TestCase):
def test_nspv_spentinfo(self): def test_nspv_spentinfo(self):
print("testing nspv_spent_info") print("testing nspv_spent_info")
result = rpc_proxy.nspv_spentinfo("67ffe0eaecd6081de04675c492a59090b573ee78955c4e8a85b8ac0be0e8e418", "1") result = rpc_proxy.nspv_spentinfo(chain_params.get(chain).get("tx_proof_id"), "1")
self.assertEqual(result["result"], "success") self.assertEqual(result["result"], "success")
self.assertEqual(result["spentheight"], 2681) self.assertEqual(result["spentheight"], chain_params.get(chain).get("tx_spent_height"))
time.sleep(1) time.sleep(1)
def test_nspv_txproof(self): def test_nspv_txproof(self):
print("testing nspv_txproof") print("testing nspv_txproof")
result = rpc_proxy.nspv_txproof("67ffe0eaecd6081de04675c492a59090b573ee78955c4e8a85b8ac0be0e8e418", "2673") result = rpc_proxy.nspv_txproof(chain_params.get(chain).get("tx_proof_id"),
self.assertEqual(result["txid"], "67ffe0eaecd6081de04675c492a59090b573ee78955c4e8a85b8ac0be0e8e418") chain_params.get(chain).get("tx_proof_height"))
self.assertEqual(result["txid"], chain_params.get(chain).get("tx_proof_id"))
time.sleep(1) time.sleep(1)
def test_nspv_login_timout(self): def test_nspv_login_timout(self):
print("testing auto-logout in 777 seconds") print("testing auto-logout in 777 seconds")
rpc_proxy.nspv_login(wif) rpc_proxy.nspv_login(wif)
time.sleep(778) time.sleep(778)
result = rpc_proxy.nspv_spend(dest_address, amount) result = rpc_proxy.nspv_spend(dest_address, amount)
self.assertEqual(result["result"], "error") self.assertEqual(result["result"], "error")
self.assertEqual(result["error"], "wif expired") self.assertEqual(result["error"], "wif expired")
time.sleep(1) time.sleep(1)
if __name__ == '__main__': if __name__ == '__main__':

20
src/tui/lib/tuilib.py
View File

@@ -21,6 +21,24 @@ if operating_system != 'Win64' and operating_system != 'Windows':
import readline import readline
class CustomProxy(Proxy):
def __init__(self,
service_url=None,
service_port=None,
conf_file=None,
timeout=3000):
config = dict()
if conf_file:
config = slickrpc.ConfigObj(conf_file)
if service_url:
config.update(self.url_to_conf(service_url))
if service_port:
config.update(rpcport=service_port)
elif not config.get('rpcport'):
config['rpcport'] = 7771
self.conn = self.prepare_connection(config, timeout=timeout)
def colorize(string, color): def colorize(string, color):
colors = { colors = {
@@ -97,7 +115,7 @@ def def_credentials(chain):
print("check "+coin_config_file) print("check "+coin_config_file)
exit(1) exit(1)
return(Proxy("http://%s:%s@127.0.0.1:%d"%(rpcuser, rpcpassword, int(rpcport)))) return(CustomProxy("http://%s:%s@127.0.0.1:%d"%(rpcuser, rpcpassword, int(rpcport))))
def getinfo_tui(rpc_connection): def getinfo_tui(rpc_connection):