tooltest.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399
  1. #!/usr/bin/env python
  2. #
  3. # mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
  4. # Copyright (C)2013-2016 Philemon <mmgen-py@yandex.com>
  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
  22. pn = os.path.dirname(sys.argv[0])
  23. os.chdir(os.path.join(pn,os.pardir))
  24. sys.path.__setitem__(0,os.path.abspath(os.curdir))
  25. # Import this _after_ local path's been added to sys.path
  26. from mmgen.common import *
  27. from collections import OrderedDict
  28. cmd_data = OrderedDict([
  29. ('util', {
  30. 'desc': 'base conversion, hashing and file utilities',
  31. 'cmd_data': OrderedDict([
  32. ('strtob58', ()),
  33. ('b58tostr', ('strtob58','io')),
  34. ('hextob58', ()),
  35. ('b58tohex', ('hextob58','io')),
  36. ('b58randenc', ()),
  37. ('hextob32', ()),
  38. ('b32tohex', ('hextob32','io')),
  39. ('randhex', ()),
  40. ('id8', ()),
  41. ('id6', ()),
  42. ('str2id6', ()),
  43. ('sha256x2', ()),
  44. ('hexreverse', ()),
  45. ('hexlify', ()),
  46. ('hexdump', ()),
  47. ('unhexdump', ('hexdump','io')),
  48. ('rand2file', ()),
  49. ])
  50. }
  51. ),
  52. ('bitcoin', {
  53. 'desc': 'Bitcoin address/key commands',
  54. 'cmd_data': OrderedDict([
  55. ('randwif', ()),
  56. ('randpair', ()),
  57. ('wif2addr', ('randpair','o2')),
  58. ('wif2hex', ('randpair','o2')),
  59. ('privhex2addr', ('wif2hex','o2')), # wif from randpair o2
  60. ('hex2wif', ('wif2hex','io2')),
  61. ('addr2hexaddr', ('randpair','o2')),
  62. ('hexaddr2addr', ('addr2hexaddr','io2')),
  63. # ('pubkey2addr', ['<public key in hex format> [str]']),
  64. # ('pubkey2hexaddr', ['<public key in hex format> [str]']),
  65. ])
  66. }
  67. ),
  68. ('mnemonic', {
  69. 'desc': 'mnemonic commands',
  70. 'cmd_data': OrderedDict([
  71. ('hex2mn', ()),
  72. ('mn2hex', ('hex2mn','io3')),
  73. ('mn_rand128', ()),
  74. ('mn_rand192', ()),
  75. ('mn_rand256', ()),
  76. ('mn_stats', ()),
  77. ('mn_printlist', ()),
  78. ])
  79. }
  80. ),
  81. ('rpc', {
  82. 'desc': 'Bitcoind RPC commands',
  83. 'cmd_data': OrderedDict([
  84. # ('keyaddrfile_chksum', ()), # interactive
  85. ('addrfile_chksum', ()),
  86. ('getbalance', ()),
  87. ('listaddresses', ()),
  88. ('twview', ()),
  89. ('txview', ()),
  90. ])
  91. }
  92. ),
  93. ])
  94. cfg = {
  95. 'name': 'the tool utility',
  96. 'enc_passwd': 'Ten Satoshis',
  97. 'tmpdir': 'test/tmp10',
  98. 'tmpdir_num': 10,
  99. 'refdir': 'test/ref',
  100. 'txfile': 'FFB367[1.234].rawtx',
  101. 'addrfile': '98831F3A[1,31-33,500-501,1010-1011].addrs',
  102. 'addrfile_chk': '6FEF 6FB9 7B13 5D91',
  103. }
  104. opts_data = {
  105. 'desc': "Test suite for the 'mmgen-tool' utility",
  106. 'usage':'[options] [command]',
  107. 'options': """
  108. -h, --help Print this help message
  109. --, --longhelp Print help message for long options (common options)
  110. -l, --list-cmds List and describe the tests and commands in the test suite
  111. -s, --system Test scripts and modules installed on system rather than
  112. those in the repo root
  113. -v, --verbose Produce more verbose output
  114. """,
  115. 'notes': """
  116. If no command is given, the whole suite of tests is run.
  117. """
  118. }
  119. sys.argv = [sys.argv[0]] + ['--skip-cfg-file'] + sys.argv[1:]
  120. cmd_args = opts.init(opts_data,add_opts=['exact_output','profile'])
  121. add_spawn_args = ' '.join(['{} {}'.format(
  122. '--'+k.replace('_','-'),
  123. getattr(opt,k) if getattr(opt,k) != True else ''
  124. ) for k in 'testnet','rpc_host' if getattr(opt,k)]).split()
  125. if opt.system: sys.path.pop(0)
  126. if opt.list_cmds:
  127. fs = ' {:<{w}} - {}'
  128. Msg('Available commands:')
  129. w = max([len(i) for i in 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()
  135. import binascii
  136. from mmgen.test import *
  137. from mmgen.tx import is_wif,is_btc_addr,is_b58_str
  138. class MMGenToolTestSuite(object):
  139. def __init__(self):
  140. pass
  141. def gen_deps_for_cmd(self,cmd,cdata):
  142. fns = []
  143. if cdata:
  144. name,code = cdata
  145. io,count = (code[:-1],int(code[-1])) if code[-1] in '0123456789' else (code,1)
  146. for c in range(count):
  147. fns += ['%s%s%s' % (
  148. name,
  149. ('',c+1)[count > 1],
  150. ('.out','.in')[ch=='i']
  151. ) for ch in io]
  152. return fns
  153. def get_num_exts_for_cmd(self,cmd,dpy): # dpy required here
  154. num = str(tool_cfgs['tmpdir_num'])
  155. # return only first file - a hack
  156. exts = gen_deps_for_cmd(dpy)
  157. return num,exts
  158. def do_cmds(self,cmd_group):
  159. cdata = cmd_data[cmd_group]['cmd_data']
  160. for cmd in cdata: self.do_cmd(cmd,cdata[cmd])
  161. def do_cmd(self,cmd,cdata):
  162. fns = self.gen_deps_for_cmd(cmd,cdata)
  163. file_list = [os.path.join(cfg['tmpdir'],fn) for fn in fns]
  164. self.__class__.__dict__[cmd](*([self,cmd] + file_list))
  165. def run_cmd(self,name,tool_args,kwargs='',extra_msg='',silent=False,strip=True):
  166. mmgen_tool = 'mmgen-tool'
  167. if not opt.system:
  168. mmgen_tool = os.path.join(os.curdir,mmgen_tool)
  169. sys_cmd = ['python',mmgen_tool] + add_spawn_args + ['-r0','-d',cfg['tmpdir'],name] + tool_args + kwargs.split()
  170. if extra_msg: extra_msg = '(%s)' % extra_msg
  171. full_name = ' '.join([name]+kwargs.split()+extra_msg.split())
  172. if not silent:
  173. if opt.verbose:
  174. sys.stderr.write(green('Testing %s\nExecuting ' % full_name))
  175. sys.stderr.write('%s\n' % cyan(repr(sys_cmd)))
  176. else:
  177. msg_r('Testing %-31s%s' % (full_name+':',''))
  178. import subprocess
  179. p = subprocess.Popen(
  180. sys_cmd,
  181. stdout=subprocess.PIPE,
  182. stderr=subprocess.PIPE,
  183. )
  184. a,b = p.communicate()
  185. retcode = p.wait()
  186. if retcode != 0:
  187. msg('%s\n%s\n%s'%(red('FAILED'),yellow('Command stderr output:'),b))
  188. die(1,red('Called process returned with an error (retcode %s)' % retcode))
  189. return (a,a.rstrip())[bool(strip)]
  190. def run_cmd_chk(self,name,f1,f2,kwargs='',extra_msg=''):
  191. idata = read_from_file(f1).rstrip()
  192. odata = read_from_file(f2).rstrip()
  193. ret = self.run_cmd(name,[odata],kwargs=kwargs,extra_msg=extra_msg)
  194. vmsg('In: ' + repr(odata))
  195. vmsg('Out: ' + repr(ret))
  196. if ret == idata: ok()
  197. else:
  198. die(3,red(
  199. "Error: values don't match:\nIn: %s\nOut: %s" % (repr(idata),repr(ret))))
  200. return ret
  201. def run_cmd_nochk(self,name,f1,kwargs=''):
  202. odata = read_from_file(f1).rstrip()
  203. ret = self.run_cmd(name,[odata],kwargs=kwargs)
  204. vmsg('In: ' + repr(odata))
  205. vmsg('Out: ' + repr(ret))
  206. return ret
  207. def run_cmd_out(self,name,carg=None,Return=False,kwargs='',fn_idx='',extra_msg='',literal=False,chkdata=''):
  208. if carg: write_to_tmpfile(cfg,'%s%s.in' % (name,fn_idx),carg+'\n')
  209. ret = self.run_cmd(name,([],[carg])[bool(carg)],kwargs=kwargs,extra_msg=extra_msg)
  210. if carg: vmsg('In: ' + repr(carg))
  211. vmsg('Out: ' + (repr(ret),ret)[literal])
  212. if ret or ret == '':
  213. write_to_tmpfile(cfg,'%s%s.out' % (name,fn_idx),ret+'\n')
  214. if chkdata:
  215. cmp_or_die(ret,chkdata)
  216. return
  217. if Return: return ret
  218. else: ok()
  219. else:
  220. die(3,red("Error for command '%s'" % name))
  221. def run_cmd_randinput(self,name,strip=True):
  222. s = os.urandom(128)
  223. fn = name+'.in'
  224. write_to_tmpfile(cfg,fn,s,binary=True)
  225. ret = self.run_cmd(name,[get_tmpfile_fn(cfg,fn)],strip=strip)
  226. fn = name+'.out'
  227. write_to_tmpfile(cfg,fn,ret+'\n')
  228. ok()
  229. vmsg('Returned: %s' % ret)
  230. def str2id6(self,name):
  231. s = getrandstr(120,no_space=True)
  232. s2 = ' %s %s %s %s %s ' % (s[:3],s[3:9],s[9:29],s[29:50],s[50:120])
  233. ret1 = self.run_cmd(name,[s],extra_msg='unspaced input'); ok()
  234. ret2 = self.run_cmd(name,[s2],extra_msg='spaced input')
  235. cmp_or_die(ret1,ret2)
  236. vmsg('Returned: %s' % ret1)
  237. def mn_rand128(self,name):
  238. self.run_cmd_out(name)
  239. def mn_rand192(self,name):
  240. self.run_cmd_out(name)
  241. def mn_rand256(self,name):
  242. self.run_cmd_out(name)
  243. def mn_stats(self,name):
  244. self.run_cmd_out(name)
  245. def mn_printlist(self,name):
  246. self.run_cmd(name,[])
  247. ok()
  248. def id6(self,name): self.run_cmd_randinput(name)
  249. def id8(self,name): self.run_cmd_randinput(name)
  250. def hexdump(self,name): self.run_cmd_randinput(name,strip=False)
  251. def unhexdump(self,name,fn1,fn2):
  252. ret = self.run_cmd(name,[fn2],strip=False)
  253. orig = read_from_file(fn1,binary=True)
  254. cmp_or_die(orig,ret)
  255. def rand2file(self,name):
  256. of = name + '.out'
  257. dlen = 1024
  258. self.run_cmd(name,[of,str(1024),'threads=4','silent=1'],strip=False)
  259. d = read_from_tmpfile(cfg,of,binary=True)
  260. cmp_or_die(dlen,len(d))
  261. def strtob58(self,name): self.run_cmd_out(name,getrandstr(16))
  262. def sha256x2(self,name): self.run_cmd_out(name,getrandstr(16))
  263. def hexreverse(self,name): self.run_cmd_out(name,getrandhex(24))
  264. def hexlify(self,name): self.run_cmd_out(name,getrandstr(24))
  265. def b58tostr(self,name,f1,f2): self.run_cmd_chk(name,f1,f2)
  266. def hextob58(self,name): self.run_cmd_out(name,getrandhex(32))
  267. def b58tohex(self,name,f1,f2): self.run_cmd_chk(name,f1,f2)
  268. def hextob32(self,name): self.run_cmd_out(name,getrandhex(24))
  269. def b32tohex(self,name,f1,f2): self.run_cmd_chk(name,f1,f2)
  270. def b58randenc(self,name):
  271. ret = self.run_cmd_out(name,Return=True)
  272. ok_or_die(ret,is_b58_str,'base 58 string')
  273. def randhex(self,name):
  274. ret = self.run_cmd_out(name,Return=True)
  275. ok_or_die(ret,binascii.unhexlify,'hex string')
  276. def randwif(self,name):
  277. for n,k in enumerate(['','compressed=1']):
  278. ret = self.run_cmd_out(name,kwargs=k,Return=True,fn_idx=n+1)
  279. ok_or_die(ret,is_wif,'WIF key')
  280. def randpair(self,name):
  281. for n,k in enumerate(['','compressed=1']):
  282. wif,addr = self.run_cmd_out(name,kwargs=k,Return=True,fn_idx=n+1).split()
  283. ok_or_die(wif,is_wif,'WIF key',skip_ok=True)
  284. ok_or_die(addr,is_btc_addr,'Bitcoin address')
  285. def hex2wif(self,name,f1,f2,f3,f4):
  286. for n,fi,fo,k in (1,f1,f2,''),(2,f3,f4,'compressed=1'):
  287. ret = self.run_cmd_chk(name,fi,fo,kwargs=k)
  288. def wif2hex(self,name,f1,f2):
  289. for n,f,k in (1,f1,''),(2,f2,'compressed=1'):
  290. wif = read_from_file(f).split()[0]
  291. self.run_cmd_out(name,wif,kwargs=k,fn_idx=n)
  292. def wif2addr(self,name,f1,f2):
  293. for n,f,k in (1,f1,''),(2,f2,'compressed=1'):
  294. wif = read_from_file(f).split()[0]
  295. self.run_cmd_out(name,wif,kwargs=k,fn_idx=n)
  296. def addr2hexaddr(self,name,f1,f2):
  297. for n,f,m in (1,f1,''),(2,f2,'from compressed'):
  298. addr = read_from_file(f).split()[-1]
  299. self.run_cmd_out(name,addr,fn_idx=n,extra_msg=m)
  300. def hexaddr2addr(self,name,f1,f2,f3,f4):
  301. for n,fi,fo,m in (1,f1,f2,''),(2,f3,f4,'from compressed'):
  302. self.run_cmd_chk(name,fi,fo,extra_msg=m)
  303. def privhex2addr(self,name,f1,f2):
  304. key1 = read_from_file(f1).rstrip()
  305. key2 = read_from_file(f2).rstrip()
  306. for n,args in enumerate([[key1],[key2,'compressed=1']]):
  307. ret = self.run_cmd(name,args).rstrip()
  308. iaddr = read_from_tmpfile(cfg,'randpair%s.out' % (n+1)).split()[-1]
  309. cmp_or_die(iaddr,ret)
  310. def hex2mn(self,name):
  311. for n,size,m in(1,16,'128-bit'),(2,24,'192-bit'),(3,32,'256-bit'):
  312. hexnum = getrandhex(size)
  313. self.run_cmd_out(name,hexnum,fn_idx=n,extra_msg=m)
  314. def mn2hex(self,name,f1,f2,f3,f4,f5,f6):
  315. for f_i,f_o,m in (f1,f2,'128-bit'),(f3,f4,'192-bit'),(f5,f6,'256-bit'):
  316. self.run_cmd_chk(name,f_i,f_o,extra_msg=m)
  317. def getbalance(self,name):
  318. self.run_cmd_out(name,literal=True)
  319. def listaddresses(self,name):
  320. self.run_cmd_out(name,literal=True)
  321. def twview(self,name):
  322. self.run_cmd_out(name,literal=True)
  323. def txview(self,name):
  324. fn = os.path.join(cfg['refdir'],cfg['txfile'])
  325. self.run_cmd_out(name,fn,literal=True)
  326. def addrfile_chksum(self,name):
  327. fn = os.path.join(cfg['refdir'],cfg['addrfile'])
  328. self.run_cmd_out(name,fn,literal=True,chkdata=cfg['addrfile_chk'])
  329. # main()
  330. import time
  331. start_time = int(time.time())
  332. ts = MMGenToolTestSuite()
  333. mk_tmpdir(cfg['tmpdir'])
  334. if cmd_args:
  335. if len(cmd_args) != 1:
  336. die(1,'Only one command may be specified')
  337. cmd = cmd_args[0]
  338. if cmd in cmd_data:
  339. msg('Running tests for %s:' % cmd_data[cmd]['desc'])
  340. ts.do_cmds(cmd)
  341. elif cmd == 'clean':
  342. cleandir(cfg['tmpdir'])
  343. sys.exit()
  344. else:
  345. die(1,"'%s': unrecognized command" % cmd)
  346. else:
  347. cleandir(cfg['tmpdir'])
  348. for cmd in cmd_data:
  349. msg('Running tests for %s:' % cmd_data[cmd]['desc'])
  350. ts.do_cmds(cmd)
  351. if cmd is not cmd_data.keys()[-1]: msg('')
  352. t = int(time.time()) - start_time
  353. msg(green(
  354. 'All requested tests finished OK, elapsed time: %02i:%02i' % (t/60,t%60)))