test scan_tx latest / oldest txid

scan_tx.py

from monerorpc.authproxy import AuthServiceProxy, JSONRPCException
import os
import pprint as pp
import sys

wallet_dir = "/some/dir"
daemon_address = ""
rpc_url = "http://127.0.0.1:18084/json_rpc"
test_wallet = ""
rpc_connection = AuthServiceProxy(service_url=rpc_url)

def get_txids_2(get_transfers):
	txids = {}
	for i in get_transfers["in"]:
		txids[i["txid"]] = i["height"]		
	for i in get_transfers["out"]:
		txids[i["txid"]] = i["height"]
	return txids

def restore_wallet(fname,seed,height,txids):
	global wallet_dir, rpc_connection
	params = {
		"filename": fname,
		"seed": seed,
		"password": "",
		"restore_height": height
	}

	if os.path.isfile(f"{wallet_dir}{fname}"):
		os.remove(f"{wallet_dir}{fname}")
		os.remove(f"{wallet_dir}{fname}.keys")

	make_new = rpc_connection.restore_deterministic_wallet(params)
	for txid in txids:
		rpc_connection.scan_tx({"txids":[txid[0]]})

def pop_it(index,old_dict):
	i=0
	new_dict = {}
	for txid in old_dict:
		if i == index:
			rejected_txid = txid[0]
			i+=1
			continue
		new_dict[str(txid[0])] = txid[1]
		i+=1
	new_dict = sorted(new_dict.items(), key=lambda x: int(x[1]))
	return (new_dict,rejected_txid)

def main(gui=False):
	global wallet_dir,rpc_url,test_wallet,daemon_address,rpc_connection
	try: rpc_connection.close_wallet()
	except: pass

	# Open an existing wallet that has 30~ transactions (in/out)
	rpc_connection.open_wallet({"filename": test_wallet,"password": ""})
	transfers = rpc_connection.get_transfers({"in": True,"out": True,})
	txids = get_txids_2(transfers)
	sorted_txids = sorted(txids.items(), key=lambda x: int(x[1]))

	seed = rpc_connection.query_key({"key_type":"mnemonic"})["key"]
	height = rpc_connection.get_height()["height"]

	data = pop_it(0,sorted_txids)
	missing_oldest = data[0]
	missing_oldest_txid = data[1]
	data = pop_it(len(sorted_txids)-1,sorted_txids)
	missing_latest = data[0]
	missing_latest_txid = data[1]

	restore_wallet("missing_oldest",seed,height,missing_oldest)
	rpc_connection.store()
	restore_wallet("missing_latest",seed,height,missing_latest)
	rpc_connection.store()
	
	if gui:
		#Open wallts in the gui and enter manually in simple mode / advanced mode with trusted/untrusted daemon
		print(f"missing_oldest: {missing_oldest_txid}")
		print(f"missing_latest: {missing_latest_txid}")
		return
	
	rpc_connection.close_wallet()

	rpc_connection.open_wallet({"filename": "missing_oldest","password": ""})
	rpc_connection.set_daemon({"address":daemon_address,"trusted":False})

	print("Untrusted: error is expected scanning oldest txid")
	try: rpc_connection.scan_tx({"txids":[missing_oldest_txid]})
	except: pass

	#Error: '{'error': {'code': -1, 'message': 'The wallet has already seen 1 or more recent transactions than the scanned tx. Either connect to a trusted daemon or rescan the chain.'}, 'id': 75, 'jsonrpc': '2.0'}'

	print("Trusted: no error expected scanning oldest txid")
	rpc_connection.set_daemon({"address":daemon_address,"trusted":True})
	try: rpc_connection.scan_tx({"txids":[missing_oldest_txid]})
	except: pass

	rpc_connection.close_wallet()

	rpc_connection.open_wallet({"filename": "missing_latest","password": ""})
	rpc_connection.set_daemon({"address":daemon_address,"trusted":False})
	print("Untrusted: no error expected scanning latest txid")
	try: rpc_connection.scan_tx({"txids":[missing_latest_txid]})
	except: pass

	rpc_connection.set_daemon({"address":daemon_address,"trusted":True})
	print("Trusted: no error expected scanning latest txid")
	try: rpc_connection.scan_tx({"txids":[missing_latest_txid]})
	except: pass

if __name__ == "__main__":
	main()

'''
Untrusted: error is expected scanning oldest txid
Error: '{'error': {'code': -1, 'message': 'The wallet has already seen 1 or more recent transactions than the scanned tx. Either connect to a trusted daemon or rescan the chain.'}, 'id': 77, 'jsonrpc': '2.0'}'
Trusted: no error expected scanning oldest txid
Untrusted: no error expected scanning latest txid
Trusted: no error expected scanning latest txid
'''