tooltest.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  1. #!/usr/bin/env python3
  2. #
  3. # mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
  4. # Copyright (C)2013-2021 The MMGen Project <mmgen@tuta.io>
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation, either version 3 of the License, or
  9. # (at your option) any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License
  17. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18. """
  19. test/tooltest.py: Tests for the 'mmgen-tool' utility
  20. """
  21. import sys,os,binascii
  22. from subprocess import run,PIPE
  23. from include.tests_header import repo_root
  24. from mmgen.common import *
  25. from test.include.common import *
  26. opts_data = {
  27. 'text': {
  28. 'desc': "Test suite for the 'mmgen-tool' utility",
  29. 'usage':'[options] [command]',
  30. 'options': """
  31. -h, --help Print this help message
  32. -C, --coverage Produce code coverage info using trace module
  33. -d, --debug Produce debugging output (stderr from spawned script)
  34. --, --longhelp Print help message for long options (common options)
  35. -l, --list-cmds List and describe the tests and commands in this test suite
  36. -L, --list-names List the names of all tested 'mmgen-tool' commands
  37. -s, --system Test scripts and modules installed on system rather than
  38. those in the repo root
  39. -t, --type=t Specify address type (valid options: 'zcash_z')
  40. -v, --verbose Produce more verbose output
  41. """,
  42. 'notes': """
  43. If no command is given, the whole suite of tests is run.
  44. """
  45. }
  46. }
  47. sys.argv = [sys.argv[0]] + ['--skip-cfg-file'] + sys.argv[1:]
  48. cmd_args = opts.init(opts_data,add_opts=['exact_output','profile'])
  49. from mmgen.protocol import init_proto_from_opts
  50. proto = init_proto_from_opts()
  51. cmd_data = {
  52. 'cryptocoin': {
  53. 'desc': 'Cryptocoin address/key commands',
  54. 'cmd_data': {
  55. 'randwif': (),
  56. 'randpair': (), # create 4 pairs: uncomp,comp,segwit,bech32
  57. 'wif2addr': ('randpair','o4'),
  58. 'wif2hex': ('randpair','o4'),
  59. 'privhex2pubhex': ('wif2hex','o3'), # segwit only
  60. 'pubhex2addr': ('privhex2pubhex','o3'), # segwit only
  61. 'hex2wif': ('wif2hex','io2'), # uncomp, comp
  62. 'addr2pubhash': ('randpair','o4'), # uncomp, comp, bech32
  63. 'pubhash2addr': ('addr2pubhash','io4'), # uncomp, comp, bech32
  64. },
  65. },
  66. 'mnemonic': {
  67. 'desc': 'mnemonic commands',
  68. 'cmd_data': {
  69. 'hex2mn': (),
  70. 'mn2hex': ('hex2mn','io3'),
  71. 'mn_rand128': (),
  72. 'mn_rand192': (),
  73. 'mn_rand256': (),
  74. 'mn_stats': (),
  75. 'mn_printlist': (),
  76. },
  77. },
  78. }
  79. if proto.coin in ('BTC','LTC'):
  80. cmd_data['cryptocoin']['cmd_data'].update({
  81. 'pubhex2redeem_script': ('privhex2pubhex','o3'),
  82. 'wif2redeem_script': ('randpair','o3'),
  83. 'wif2segwit_pair': ('randpair','o2'),
  84. 'privhex2addr': ('wif2hex','o4'), # compare with output of randpair
  85. 'pipetest': ('randpair','o3')
  86. })
  87. if opt.type == 'zcash_z':
  88. del cmd_data['cryptocoin']['cmd_data']['pubhash2addr']
  89. cfg = {
  90. 'name': 'the tool utility',
  91. 'enc_passwd': 'Ten Satoshis',
  92. 'tmpdir': 'test/tmp10',
  93. 'tmpdir_num': 10,
  94. 'refdir': 'test/ref',
  95. 'txfile': {
  96. 'btc': ('0B8D5A[15.31789,14,tl=1320969600].rawtx',
  97. '0C7115[15.86255,14,tl=1320969600].testnet.rawtx'),
  98. 'bch': ('460D4D-BCH[10.19764,tl=1320969600].rawtx',
  99. '359FD5-BCH[6.68868,tl=1320969600].testnet.rawtx'),
  100. 'ltc': ('AF3CDF-LTC[620.76194,1453,tl=1320969600].rawtx',
  101. 'A5A1E0-LTC[1454.64322,1453,tl=1320969600].testnet.rawtx'),
  102. },
  103. 'addrfile': '98831F3A{}[1,31-33,500-501,1010-1011]{}.addrs',
  104. 'addrfile_chk': {
  105. 'btc': ('6FEF 6FB9 7B13 5D91','424E 4326 CFFE 5F51'),
  106. 'bch': ('6FEF 6FB9 7B13 5D91','424E 4326 CFFE 5F51'),
  107. 'ltc': ('AD52 C3FE 8924 AAF0','4EBE 2E85 E969 1B30'),
  108. }
  109. }
  110. ref_subdir = '' if proto.base_coin == 'BTC' else proto.name.lower()
  111. altcoin_pfx = '' if proto.base_coin == 'BTC' else '-'+proto.base_coin
  112. tn_ext = ('','.testnet')[proto.testnet]
  113. mmgen_cmd = 'mmgen-tool'
  114. if not opt.system:
  115. os.environ['PYTHONPATH'] = repo_root
  116. mmgen_cmd = os.path.relpath(os.path.join(repo_root,'cmds',mmgen_cmd))
  117. spawn_cmd = [mmgen_cmd]
  118. if opt.coverage:
  119. d,f = init_coverage()
  120. spawn_cmd = ['python3','-m','trace','--count','--coverdir='+d,'--file='+f] + spawn_cmd
  121. elif g.platform == 'win':
  122. spawn_cmd = ['python3'] + spawn_cmd
  123. add_spawn_args = ['--data-dir='+cfg['tmpdir']] + ['--{}{}'.format(
  124. k.replace('_','-'),'='+getattr(opt,k) if getattr(opt,k) != True else '')
  125. for k in ('testnet','rpc_host','regtest','coin','type') if getattr(opt,k)]
  126. if opt.list_cmds:
  127. fs = ' {:<{w}} - {}'
  128. Msg('Available commands:')
  129. w = max(map(len,cmd_data))
  130. for cmd in cmd_data:
  131. Msg(fs.format(cmd,cmd_data[cmd]['desc'],w=w))
  132. Msg('\nAvailable utilities:')
  133. Msg(fs.format('clean','Clean the tmp directory',w=w))
  134. sys.exit(0)
  135. if opt.list_names:
  136. tcmd = ['python3','test/tooltest2.py','--list-tested-cmds']
  137. tested_in = {
  138. 'tooltest.py': [],
  139. 'test.py': (
  140. 'encrypt','decrypt','find_incog_data',
  141. 'addrfile_chksum','keyaddrfile_chksum','passwdfile_chksum',
  142. 'add_label','remove_label','remove_address','twview',
  143. 'getbalance','listaddresses','listaddress'),
  144. 'tooltest2.py': run(tcmd,stdout=PIPE,check=True).stdout.decode().split()
  145. }
  146. for v in cmd_data.values():
  147. tested_in['tooltest.py'] += list(v['cmd_data'].keys())
  148. msg(green("TESTED 'MMGEN-TOOL' COMMANDS"))
  149. for l in ('tooltest.py','tooltest2.py','test.py','test-release.sh'):
  150. msg(blue(l+':'))
  151. msg(' '+'\n '.join(sorted(tested_in[l])))
  152. ignore = ()
  153. from mmgen.tool import MMGenToolCmd
  154. uc = sorted(
  155. set(MMGenToolCmds) -
  156. set(ignore) -
  157. set(tested_in['tooltest.py']) -
  158. set(tested_in['tooltest2.py']) -
  159. set(tested_in['test.py']) -
  160. set(tested_in['test-release.sh'])
  161. )
  162. die(0,'\n{}\n {}'.format(yellow('Untested commands:'),'\n '.join(uc)))
  163. from mmgen.tx import is_wif,is_coin_addr
  164. def is_wif_loc(s):
  165. return is_wif(proto,s)
  166. def is_coin_addr_loc(s):
  167. return is_coin_addr(proto,s)
  168. msg_w = 35
  169. def test_msg(m):
  170. m2 = 'Testing {}'.format(m)
  171. msg_r(green(m2+'\n') if opt.verbose else '{:{w}}'.format(m2,w=msg_w+8))
  172. compressed = ('','compressed')['C' in proto.mmtypes]
  173. segwit = ('','segwit')['S' in proto.mmtypes]
  174. bech32 = ('','bech32')['B' in proto.mmtypes]
  175. type_compressed_arg = ([],['--type=compressed'])['C' in proto.mmtypes]
  176. type_segwit_arg = ([],['--type=segwit'])['S' in proto.mmtypes]
  177. type_bech32_arg = ([],['--type=bech32'])['B' in proto.mmtypes]
  178. class MMGenToolTestUtils(object):
  179. def run_cmd(self,name,tool_args,kwargs='',extra_msg='',silent=False,strip=True,add_opts=[],binary=False):
  180. sys_cmd = (
  181. spawn_cmd +
  182. add_spawn_args +
  183. ['-r0','-d',cfg['tmpdir']] +
  184. add_opts +
  185. [name.lower()] +
  186. tool_args +
  187. kwargs.split()
  188. )
  189. if extra_msg: extra_msg = '({})'.format(extra_msg)
  190. full_name = ' '.join([name.lower()]+add_opts+kwargs.split()+extra_msg.split())
  191. if not silent:
  192. if opt.verbose:
  193. sys.stderr.write(green('Testing {}\nExecuting '.format(full_name)))
  194. sys.stderr.write(cyan(' '.join(sys_cmd)+'\n'))
  195. else:
  196. msg_r('Testing {:{w}}'.format(full_name+':',w=msg_w))
  197. cp = run(sys_cmd,stdout=PIPE,stderr=PIPE)
  198. out = cp.stdout
  199. err = cp.stderr
  200. if opt.debug:
  201. try: dmsg(err.decode())
  202. except: dmsg(repr(err))
  203. if not binary:
  204. out = out.decode()
  205. if cp.returncode != 0:
  206. msg('{}\n{}\n{}'.format(red('FAILED'),yellow('Command stderr output:'),err.decode()))
  207. rdie(1,'Called process returned with an error (retcode {})'.format(cp.returncode))
  208. return (out,out.rstrip())[bool(strip)]
  209. def run_cmd_chk(self,name,f1,f2,kwargs='',extra_msg='',strip_hex=False,add_opts=[]):
  210. idata = read_from_file(f1).rstrip()
  211. odata = read_from_file(f2).rstrip()
  212. ret = self.run_cmd(name,[odata],kwargs=kwargs,extra_msg=extra_msg,add_opts=add_opts)
  213. vmsg('In: ' + repr(odata))
  214. vmsg('Out: ' + repr(ret))
  215. def cmp_equal(a,b):
  216. return (a.lstrip('0') == b.lstrip('0')) if strip_hex else (a == b)
  217. if cmp_equal(ret,idata): ok()
  218. else:
  219. fs = "Error: values don't match:\nIn: {!r}\nOut: {!r}"
  220. rdie(3,fs.format(idata,ret))
  221. return ret
  222. def run_cmd_nochk(self,name,f1,kwargs='',add_opts=[]):
  223. odata = read_from_file(f1).rstrip()
  224. ret = self.run_cmd(name,[odata],kwargs=kwargs,add_opts=add_opts)
  225. vmsg('In: ' + repr(odata))
  226. vmsg('Out: ' + repr(ret))
  227. return ret
  228. def run_cmd_out(self,name,carg=None,Return=False,kwargs='',fn_idx='',extra_msg='',
  229. literal=False,chkdata='',hush=False,add_opts=[]):
  230. if carg: write_to_tmpfile(cfg,'{}{}.in'.format(name,fn_idx),carg+'\n')
  231. ret = self.run_cmd(name,([],[carg])[bool(carg)],kwargs=kwargs,
  232. extra_msg=extra_msg,add_opts=add_opts)
  233. if carg: vmsg('In: ' + repr(carg))
  234. vmsg('Out: ' + (repr(ret),ret)[literal])
  235. if ret or ret == '':
  236. write_to_tmpfile(cfg,'{}{}.out'.format(name,fn_idx),ret+'\n')
  237. if chkdata:
  238. cmp_or_die(ret,chkdata)
  239. return
  240. if Return: return ret
  241. else:
  242. if not hush: ok()
  243. else:
  244. rdie(3,"Error for command '{}'".format(name))
  245. def run_cmd_randinput(self,name,strip=True,add_opts=[]):
  246. s = os.urandom(128)
  247. fn = name+'.in'
  248. write_to_tmpfile(cfg,fn,s,binary=True)
  249. ret = self.run_cmd(name,[get_tmpfile(cfg,fn)],strip=strip,add_opts=add_opts)
  250. fn = name+'.out'
  251. write_to_tmpfile(cfg,fn,ret+'\n')
  252. ok()
  253. vmsg('Returned: {}'.format(ret))
  254. tu = MMGenToolTestUtils()
  255. def ok_or_die(val,chk_func,s,skip_ok=False):
  256. try: ret = chk_func(val)
  257. except: ret = False
  258. if ret:
  259. if not skip_ok: ok()
  260. else:
  261. rdie(3,"Returned value '{}' is not a {}".format((val,s)))
  262. class MMGenToolTestCmds(object):
  263. # Cryptocoin
  264. def randwif(self,name):
  265. for n,k in enumerate(['',compressed]):
  266. ao = ['--type='+k] if k else []
  267. ret = tu.run_cmd_out(name,add_opts=ao,Return=True,fn_idx=n+1)
  268. ok_or_die(ret,is_wif_loc,'WIF key')
  269. def randpair(self,name):
  270. for n,k in enumerate(['',compressed,segwit,bech32]):
  271. ao = ['--type='+k] if k else []
  272. wif,addr = tu.run_cmd_out(name,add_opts=ao,Return=True,fn_idx=n+1,literal=True).split()
  273. ok_or_die(wif,is_wif_loc,'WIF key',skip_ok=True)
  274. ok_or_die(addr,is_coin_addr_loc,'Coin address')
  275. def wif2addr(self,name,f1,f2,f3,f4):
  276. for n,f,k in (
  277. (1,f1,''),
  278. (2,f2,compressed),
  279. (3,f3,segwit),
  280. (4,f4,bech32)
  281. ):
  282. ao = ['--type='+k] if k else []
  283. wif = read_from_file(f).split()[0]
  284. tu.run_cmd_out(name,wif,add_opts=ao,fn_idx=n)
  285. def wif2hex(self,name,f1,f2,f3,f4):
  286. for n,f,m in (
  287. (1,f1,''),
  288. (2,f2,compressed),
  289. (3,f3,'{} for {}'.format(compressed or 'uncompressed',segwit or 'p2pkh')),
  290. (4,f4,'{} for {}'.format(compressed or 'uncompressed',bech32 or 'p2pkh'))
  291. ):
  292. wif = read_from_file(f).split()[0]
  293. tu.run_cmd_out(name,wif,fn_idx=n,extra_msg=m)
  294. def privhex2addr(self,name,f1,f2,f3,f4):
  295. keys = [read_from_file(f).rstrip() for f in (f1,f2,f3,f4)]
  296. for n,k in enumerate(('',compressed,segwit,bech32)):
  297. ao = ['--type='+k] if k else []
  298. ret = tu.run_cmd(name,[keys[n]],add_opts=ao).rstrip()
  299. iaddr = read_from_tmpfile(cfg,'randpair{}.out'.format(n+1)).split()[-1]
  300. vmsg('Out: {}'.format(ret))
  301. cmp_or_die(iaddr,ret)
  302. ok()
  303. def hex2wif(self,name,f1,f2,f3,f4):
  304. for n,fi,fo,k in ((1,f1,f2,''),(2,f3,f4,compressed)):
  305. ao = ['--type='+k] if k else []
  306. ret = tu.run_cmd_chk(name,fi,fo,add_opts=ao)
  307. def addr2pubhash(self,name,f1,f2,f3,f4):
  308. for n,f,m,ao in (
  309. (1,f1,'',[]),
  310. (2,f2,'from {}'.format(compressed or 'uncompressed'),[]),
  311. (4,f4,'',type_bech32_arg),
  312. ):
  313. addr = read_from_file(f).split()[-1]
  314. tu.run_cmd_out(name,addr,fn_idx=n,add_opts=ao,extra_msg=m)
  315. def pubhash2addr(self,name,f1,f2,f3,f4,f5,f6,f7,f8):
  316. for n,fi,fo,m,ao in (
  317. (1,f1,f2,'',[]),
  318. (2,f3,f4,'from {}'.format(compressed or 'uncompressed'),[]),
  319. (4,f7,f8,'',type_bech32_arg)
  320. ):
  321. tu.run_cmd_chk(name,fi,fo,add_opts=ao,extra_msg=m)
  322. def privhex2pubhex(self,name,f1,f2,f3): # from Hex2wif
  323. addr = read_from_file(f3).strip()
  324. tu.run_cmd_out(name,addr,add_opts=type_compressed_arg,fn_idx=3) # what about uncompressed?
  325. def pubhex2redeem_script(self,name,f1,f2,f3): # from above
  326. addr = read_from_file(f3).strip()
  327. tu.run_cmd_out(name,addr,add_opts=type_segwit_arg,fn_idx=3)
  328. rs = read_from_tmpfile(cfg,'privhex2pubhex3.out').strip()
  329. tu.run_cmd_out('pubhex2addr',rs,add_opts=type_segwit_arg,fn_idx=3,hush=True)
  330. addr1 = read_from_tmpfile(cfg,'pubhex2addr3.out').strip()
  331. addr2 = read_from_tmpfile(cfg,'randpair3.out').split()[1]
  332. cmp_or_die(addr1,addr2)
  333. ok()
  334. def wif2redeem_script(self,name,f1,f2,f3): # compare output with above
  335. wif = read_from_file(f3).split()[0]
  336. ret1 = tu.run_cmd_out(name,wif,add_opts=type_segwit_arg,fn_idx=3,Return=True)
  337. ret2 = read_from_tmpfile(cfg,'pubhex2redeem_script3.out').strip()
  338. cmp_or_die(ret1,ret2)
  339. ok()
  340. def wif2segwit_pair(self,name,f1,f2): # does its own checking, so just run
  341. wif = read_from_file(f2).split()[0]
  342. tu.run_cmd_out(name,wif,add_opts=type_segwit_arg,fn_idx=2)
  343. def pubhex2addr(self,name,f1,f2,f3):
  344. addr = read_from_file(f3).strip()
  345. tu.run_cmd_out(name,addr,add_opts=type_segwit_arg,fn_idx=3)
  346. def pipetest(self,name,f1,f2,f3):
  347. wif = read_from_file(f3).split()[0]
  348. cmd = ( '{c} {a} wif2hex {wif}' +
  349. ' | {c} {a} --type=compressed privhex2pubhex -' +
  350. ' | {c} {a} --type=segwit pubhex2redeem_script -' +
  351. ' | {c} {a} hash160 -' +
  352. ' | {c} {a} --type=segwit pubhash2addr -').format(
  353. c=' '.join(spawn_cmd),
  354. a=' '.join(add_spawn_args),
  355. wif=wif)
  356. test_msg('command piping')
  357. if opt.verbose:
  358. sys.stderr.write(green('Executing ') + cyan(cmd) + '\n')
  359. res = run(cmd,stdout=PIPE,shell=True).stdout.decode().strip()
  360. addr = read_from_tmpfile(cfg,'wif2addr3.out').strip()
  361. cmp_or_die(addr,res)
  362. ok()
  363. # Mnemonic
  364. def hex2mn(self,name):
  365. for n,size,m in ((1,16,'128-bit'),(2,24,'192-bit'),(3,32,'256-bit')):
  366. hexnum = getrandhex(size)
  367. tu.run_cmd_out(name,hexnum,fn_idx=n,extra_msg=m)
  368. def mn2hex(self,name,f1,f2,f3,f4,f5,f6):
  369. for f_i,f_o,m in ((f1,f2,'128-bit'),(f3,f4,'192-bit'),(f5,f6,'256-bit')):
  370. tu.run_cmd_chk(name,f_i,f_o,extra_msg=m,strip_hex=True)
  371. def mn_rand128(self,name): tu.run_cmd_out(name)
  372. def mn_rand192(self,name): tu.run_cmd_out(name)
  373. def mn_rand256(self,name): tu.run_cmd_out(name)
  374. def mn_stats(self,name): tu.run_cmd_out(name)
  375. def mn_printlist(self,name):
  376. tu.run_cmd(name,[])
  377. ok()
  378. # main()
  379. import time
  380. start_time = int(time.time())
  381. mk_tmpdir(cfg['tmpdir'])
  382. def gen_deps_for_cmd(cmd,cdata):
  383. fns = []
  384. if cdata:
  385. name,code = cdata
  386. io,count = (code[:-1],int(code[-1])) if code[-1] in '0123456789' else (code,1)
  387. for c in range(count):
  388. fns += ['{}{}{}'.format(name,('',c+1)[count > 1],('.out','.in')[ch=='i']) for ch in io]
  389. return fns
  390. def do_cmds(cmd_group):
  391. tc = MMGenToolTestCmds()
  392. gdata = cmd_data[cmd_group]['cmd_data']
  393. for cmd in gdata:
  394. fns = gen_deps_for_cmd(cmd,gdata[cmd])
  395. cmdline = [cmd] + [os.path.join(cfg['tmpdir'],fn) for fn in fns]
  396. getattr(tc,cmd)(*cmdline)
  397. try:
  398. if cmd_args:
  399. if len(cmd_args) != 1:
  400. die(1,'Only one command may be specified')
  401. cmd = cmd_args[0]
  402. if cmd in cmd_data:
  403. cleandir(cfg['tmpdir'],do_msg=True)
  404. msg('Running tests for {}:'.format(cmd_data[cmd]['desc']))
  405. do_cmds(cmd)
  406. elif cmd == 'clean':
  407. cleandir(cfg['tmpdir'],do_msg=True)
  408. sys.exit(0)
  409. else:
  410. die(1,"'{}': unrecognized command".format(cmd))
  411. else:
  412. cleandir(cfg['tmpdir'],do_msg=True)
  413. for cmd in cmd_data:
  414. msg('Running tests for {}:'.format(cmd_data[cmd]['desc']))
  415. do_cmds(cmd)
  416. if cmd is not list(cmd_data.keys())[-1]: msg('')
  417. except KeyboardInterrupt:
  418. die(1,green('\nExiting at user request'))
  419. t = int(time.time()) - start_time
  420. gmsg('All requested tests finished OK, elapsed time: {:02}:{:02}'.format(t//60,t%60))