#!/usr/bin/env python3 # # mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution # Copyright (C)2013-2019 The MMGen Project # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . """ test/test.py: Test suite for the MMGen wallet system """ def check_segwit_opts(): for k,m in (('segwit','S'),('segwit_random','S'),('bech32','B')): if getattr(opt,k) and m not in g.proto.mmtypes: die(1,'--{} option incompatible with {}'.format(k.replace('_','-'),g.proto.__name__)) def create_shm_dir(data_dir,trash_dir): # Laggy flash media can cause pexpect to fail, so create a temporary directory # under '/dev/shm' and put datadir and tmpdirs here. import shutil if g.platform == 'win': for tdir in (data_dir,trash_dir): try: os.listdir(tdir) except: pass else: try: shutil.rmtree(tdir) except: # we couldn't remove data dir - perhaps regtest daemon is running try: subprocess.call(['python3',os.path.join('cmds','mmgen-regtest'),'stop']) except: rdie(1,"Unable to remove {!r}!".format(tdir)) else: time.sleep(2) shutil.rmtree(tdir) os.mkdir(tdir,0o755) shm_dir = 'test' else: tdir,pfx = '/dev/shm','mmgen-test-' try: subprocess.call('rm -rf {}/{}*'.format(tdir,pfx),shell=True) except Exception as e: die(2,'Unable to delete directory tree {}/{}* ({})'.format(tdir,pfx,e.args[0])) try: import tempfile shm_dir = str(tempfile.mkdtemp('',pfx,tdir)) except Exception as e: die(2,'Unable to create temporary directory in {} ({})'.format(tdir,e.args[0])) dest = os.path.join(shm_dir,os.path.basename(trash_dir)) os.mkdir(dest,0o755) try: os.unlink(trash_dir) except: pass os.symlink(dest,trash_dir) dest = os.path.join(shm_dir,os.path.basename(data_dir)) shutil.move(data_dir,dest) # data_dir was created by opts.init() os.symlink(dest,data_dir) return shm_dir import sys,os,time repo_root = os.path.normpath(os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]),os.pardir))) os.chdir(repo_root) sys.path.__setitem__(0,repo_root) try: os.unlink(os.path.join(repo_root,'my.err')) except: pass # Import these _after_ local path's been added to sys.path from mmgen.common import * from test.common import * from test.test_py_d.common import * g.quiet = False # if 'quiet' was set in config file, disable here os.environ['MMGEN_QUIET'] = '0' # for this script and spawned scripts opts_data = { 'text': { 'desc': 'Test suite for the MMGen suite', 'usage':'[options] [command(s) or metacommand(s)]', 'options': """ -h, --help Print this help message --, --longhelp Print help message for long options (common options) -B, --bech32 Generate and use Bech32 addresses -b, --buf-keypress Use buffered keypresses as with real human input (often required on slow systems, or under emulation) -c, --print-cmdline Print the command line of each spawned command -C, --coverage Produce code coverage info using trace module -x, --debug-pexpect Produce debugging output for pexpect calls -D, --no-daemon-stop Don't stop auto-started daemons after running tests -E, --direct-exec Bypass pexpect and execute a command directly (for debugging only) -e, --exact-output Show the exact output of the MMGen script(s) being run -G, --exclude-groups=G Exclude the specified command groups (comma-separated) -l, --list-cmds List and describe the commands in the test suite -L, --list-cmd-groups Output a list of command groups, with no descriptions -n, --names Display command names instead of descriptions -o, --log Log commands to file {lf} -O, --pexpect-spawn Use pexpect.spawn instead of popen_spawn (much slower, kut does real terminal emulation) -p, --pause Pause between tests, resuming on keypress -P, --profile Record the execution time of each script -q, --quiet Produce minimal output. Suppress dependency info -r, --resume=c Resume at command 'c' after interrupted run -s, --system Test scripts and modules installed on system rather than those in the repo root -S, --skip-deps Skip dependency checking for command -u, --usr-random Get random data interactively from user -t, --traceback Run the command inside the '{tbc}' script -T, --pexpect-timeout=T Set the timeout for pexpect -v, --verbose Produce more verbose output -W, --no-dw-delete Don't remove default wallet from data dir after dw tests are done -X, --exit-after=C Exit after command 'C' -y, --segwit Generate and use Segwit addresses -Y, --segwit-random Generate and use a random mix of Segwit and Legacy addrs """, 'notes': """ If no command is given, the whole test suite is run. """ }, 'code': { 'options': lambda s: s.format( tbc='scripts/traceback_run.py', lf=log_file), } } data_dir = os.path.join('test','data_dir' + ('','-α')[bool(os.getenv('MMGEN_DEBUG_UTF8'))]) # we need the values of two opts before running opts.init, so parse without initializing: uopts = opts.init(opts_data,parse_only=True)[0] # step 1: delete data_dir symlink in ./test; if not ('resume' in uopts or 'skip_deps' in uopts): try: os.unlink(data_dir) except: pass sys.argv = [sys.argv[0]] + ['--data-dir='+data_dir] + sys.argv[1:] # step 2: opts.init will create new data_dir in ./test (if not 'resume' or 'skip_deps'): usr_args = opts.init(opts_data) # step 3: move data_dir to /dev/shm and symlink it back to ./test: trash_dir = os.path.join('test','trash') if not ('resume' in uopts or 'skip_deps' in uopts): shm_dir = create_shm_dir(data_dir,trash_dir) check_segwit_opts() if opt.profile: opt.names = True if opt.resume: opt.skip_deps = True if opt.exact_output: def msg(s): pass qmsg = qmsg_r = vmsg = vmsg_r = msg_r = msg cfgs = { # addr_idx_lists (except 31,32,33,34) must contain exactly 8 addresses '1': { 'wpasswd': 'Dorian-α', 'kapasswd': 'Grok the blockchain', 'addr_idx_list': '12,99,5-10,5,12', 'dep_generators': { pwfile: 'walletgen', 'mmdat': 'walletgen', 'addrs': 'addrgen', 'rawtx': 'txcreate', 'txbump': 'txbump', 'sigtx': 'txsign', 'mmwords': 'export_mnemonic', 'mmseed': 'export_seed', 'mmhex': 'export_hex', 'mmincog': 'export_incog', 'mmincox': 'export_incog_hex', hincog_fn: 'export_incog_hidden', incog_id_fn: 'export_incog_hidden', 'akeys.mmenc': 'keyaddrgen' }, }, '2': { 'wpasswd': 'Hodling away', 'addr_idx_list': '37,45,3-6,22-23', 'seed_len': 128, 'dep_generators': { 'mmdat': 'walletgen2', 'addrs': 'addrgen2', 'rawtx': 'txcreate2', 'sigtx': 'txsign2', 'mmwords': 'export_mnemonic2', }, }, '3': { 'wpasswd': 'Major miner', 'addr_idx_list': '73,54,1022-1023,2-5', 'dep_generators': { 'mmdat': 'walletgen3', 'addrs': 'addrgen3', 'rawtx': 'txcreate3', 'sigtx': 'txsign3' }, }, '4': { 'wpasswd': 'Hashrate good', 'addr_idx_list': '63,1004,542-544,7-9', 'seed_len': 192, 'dep_generators': { 'mmdat': 'walletgen4', 'mmbrain': 'walletgen4', 'addrs': 'addrgen4', 'rawtx': 'txcreate4', 'sigtx': 'txsign4', 'txdo': 'txdo4', }, 'bw_filename': 'brainwallet.mmbrain', 'bw_params': '192,1', }, '5': { 'wpasswd': 'My changed password-α', 'hash_preset': '2', 'dep_generators': { 'mmdat': 'passchg', pwfile: 'passchg', }, }, '6': { 'seed_len': 128, 'seed_id': 'FE3C6545', 'ref_bw_seed_id': '33F10310', 'wpasswd': 'reference password', 'kapasswd': '', 'dep_generators': { 'mmdat': 'refwalletgen_1', pwfile: 'refwalletgen_1', 'addrs': 'refaddrgen_1', 'akeys.mmenc': 'refkeyaddrgen_1' }, }, '7': { 'seed_len': 192, 'seed_id': '1378FC64', 'ref_bw_seed_id': 'CE918388', 'wpasswd': 'reference password', 'kapasswd': '', 'dep_generators': { 'mmdat': 'refwalletgen_2', pwfile: 'refwalletgen_2', 'addrs': 'refaddrgen_2', 'akeys.mmenc': 'refkeyaddrgen_2' }, }, '8': { 'seed_len': 256, 'seed_id': '98831F3A', 'ref_bw_seed_id': 'B48CD7FC', 'wpasswd': 'reference password', 'kapasswd': '', 'dep_generators': { 'mmdat': 'refwalletgen_3', pwfile: 'refwalletgen_3', 'addrs': 'refaddrgen_3', 'akeys.mmenc': 'refkeyaddrgen_3' }, }, '9': { 'tool_enc_infn': 'tool_encrypt.in', 'dep_generators': { 'tool_encrypt.in': 'tool_encrypt', 'tool_encrypt.in.mmenc': 'tool_encrypt', }, }, '14': { 'kapasswd': 'Maxwell', 'wpasswd': 'The Halving', 'addr_idx_list': '61,998,502-504,7-9', 'seed_len': 256, 'dep_generators': { 'mmdat': 'walletgen14', 'addrs': 'addrgen14', 'akeys.mmenc': 'keyaddrgen14', }, }, '15': { 'wpasswd': 'Dorian-α', 'kapasswd': 'Grok the blockchain', 'addr_idx_list': '12,99,5-10,5,12', 'dep_generators': { pwfile: 'walletgen_dfl_wallet', 'addrs': 'addrgen_dfl_wallet', 'rawtx': 'txcreate_dfl_wallet', 'sigtx': 'txsign_dfl_wallet', 'mmseed': 'export_seed_dfl_wallet', 'del_dw_run': 'delete_dfl_wallet', }, }, '16': { 'wpasswd': 'My changed password', 'hash_preset': '2', 'dep_generators': { pwfile: 'passchg_dfl_wallet', }, }, '17': {}, '18': {}, '19': { 'wpasswd':'abc' }, # B2X '20': { 'wpasswd': 'Vsize it', 'addr_idx_list': '1-8', 'seed_len': 256, 'dep_generators': { 'mmdat': 'walletgen5', 'addrs': 'addrgen5', 'rawtx': 'txcreate5', 'sigtx': 'txsign5', }, }, '21': { 'wpasswd': 'Vsize it', 'addr_idx_list': '1-8', 'seed_len': 256, 'dep_generators': { 'mmdat': 'walletgen6', 'addrs': 'addrgen6', 'rawtx': 'txcreate6', 'sigtx': 'txsign6', }, }, '22': {}, '31': {}, '32': {}, '33': {}, '34': {}, } for k in cfgs: cfgs[k]['tmpdir'] = os.path.join('test','tmp{}'.format(k)) cfgs[k]['segwit'] = randbool() if opt.segwit_random else bool(opt.segwit or opt.bech32) from copy import deepcopy for a,b in (('6','11'),('7','12'),('8','13')): cfgs[b] = deepcopy(cfgs[a]) cfgs[b]['tmpdir'] = os.path.join('test','tmp'+b) if g.debug_utf8: for k in cfgs: cfgs[k]['tmpdir'] += '-α' utils = { # 'check_deps': 'check dependencies for specified command (WIP)', # TODO 'clean': 'clean specified tmp dir(s) (specify by integer, no arg = all dirs)', } def list_cmds(): gm = CmdGroupMgr() cw,d = 0,[] Msg(green('AVAILABLE COMMANDS:')) for gname in gm.cmd_groups: ts = gm.init_group(None,gname,None) d.append((gname,ts.__doc__.strip(),gm.cmd_list,gm.dpy_data)) cw = max(max(len(k) for k in gm.dpy_data),cw) for gname,gdesc,clist,dpdata in d: Msg('\n'+green('{!r} - {}:'.format(gname,gdesc))) for cmd in clist: data = dpdata[cmd] Msg(' {:{w}} - {}'.format(cmd,data if type(data) == str else data[1],w=cw)) w = max(map(len,utils)) Msg('\n'+green('AVAILABLE UTILITIES:')) for cmd in sorted(utils): Msg(' {:{w}} - {}'.format(cmd,utils[cmd],w=w)) sys.exit(0) def do_between(): if opt.pause: confirm_continue() elif (opt.verbose or opt.exact_output) and not opt.skip_deps: sys.stderr.write('\n') def list_tmpdirs(): return {k:cfgs[k]['tmpdir'] for k in cfgs} def clean(usr_dirs=None): if opt.skip_deps: return all_dirs = list_tmpdirs() dirnums = map(int,(usr_dirs if usr_dirs is not None else all_dirs)) dirlist = list(map(str,sorted(dirnums))) for d in dirlist: if d in all_dirs: cleandir(all_dirs[d]) else: die(1,'{}: invalid directory number'.format(d)) if dirlist: iqmsg(green('Cleaned tmp director{} {}'.format(suf(dirlist,'y'),' '.join(dirlist)))) cleandir(data_dir) cleandir(trash_dir) iqmsg(green("Cleaned directories '{}'".format("' '".join([data_dir,trash_dir])))) def create_tmp_dirs(shm_dir): if g.platform == 'win': for cfg in sorted(cfgs): mk_tmpdir(cfgs[cfg]['tmpdir']) else: for cfg in sorted(cfgs): src = os.path.join(shm_dir,cfgs[cfg]['tmpdir'].split('/')[-1]) mk_tmpdir(src) try: os.unlink(cfgs[cfg]['tmpdir']) except OSError as e: if e.errno != 2: raise finally: os.symlink(src,cfgs[cfg]['tmpdir']) def set_environ_for_spawned_scripts(): if os.getenv('MMGEN_DEBUG_ALL'): for name in g.env_opts: if name[:11] == 'MMGEN_DEBUG': os.environ[name] = '1' if not opt.pexpect_spawn: os.environ['MMGEN_TEST_SUITE_POPEN_SPAWN'] = '1' if not opt.system: os.environ['PYTHONPATH'] = repo_root if not opt.buf_keypress: os.environ['MMGEN_DISABLE_HOLD_PROTECT'] = '1' # If test.py itself is running under traceback, the spawned script shouldn't be, so disable this: if os.getenv('MMGEN_TRACEBACK') and not opt.traceback: os.environ['MMGEN_TRACEBACK'] = '' # Disable color in spawned scripts so pexpect can parse their output os.environ['MMGEN_DISABLE_COLOR'] = '1' os.environ['MMGEN_NO_LICENSE'] = '1' os.environ['MMGEN_MIN_URANDCHARS'] = '3' os.environ['MMGEN_BOGUS_SEND'] = '1' # Tell spawned programs they're running in the test suite os.environ['MMGEN_TEST_SUITE'] = '1' def set_restore_term_at_exit(): import termios,atexit fd = sys.stdin.fileno() old = termios.tcgetattr(fd) def at_exit(): termios.tcsetattr(fd, termios.TCSADRAIN, old) atexit.register(at_exit) class CmdGroupMgr(object): cmd_groups = { 'helpscreens': ('TestSuiteHelp',{'modname':'misc','full_data':True}), 'main': ('TestSuiteMain',{'full_data':True}), 'conv': ('TestSuiteWalletConv',{'is3seed':True,'modname':'wallet'}), 'ref3': ('TestSuiteRef3Seed',{'is3seed':True,'modname':'ref_3seed'}), 'ref': ('TestSuiteRef',{}), 'ref_altcoin': ('TestSuiteRefAltcoin',{}), 'tool': ('TestSuiteTool',{'modname':'misc','full_data':True}), 'regtest': ('TestSuiteRegtest',{}), # 'chainsplit': ('TestSuiteChainsplit',{}), 'ethdev': ('TestSuiteEthdev',{}), 'autosign': ('TestSuiteAutosign',{}), 'autosign_minimal': ('TestSuiteAutosignMinimal',{'modname':'autosign'}), 'autosign_live': ('TestSuiteAutosignLive',{'modname':'autosign'}), 'create_ref_tx': ('TestSuiteRefTX',{'modname':'misc','full_data':True}), } dfl_groups = ( 'helpscreens', 'main', 'conv', 'ref', 'ref3', 'ref_altcoin', 'tool', 'autosign_minimal', 'regtest', 'ethdev') def load_mod(self,gname,modname=None): clsname,kwargs = self.cmd_groups[gname] if modname == None and 'modname' in kwargs: modname = kwargs['modname'] gl = globals() exec('from test.test_py_d import ts_{}'.format(modname or gname),gl,gl) exec('from test.test_py_d.ts_{} import {}'.format(modname or gname,clsname),gl,gl) return clsname def create_group(self,gname,full_data=False,modname=None,is3seed=False,add_dpy=False): """ Initializes the list 'cmd_list' and dict 'dpy_data' from module's cmd_group data. Alternatively, if called with 'add_dpy=True', updates 'dpy_data' from module data without touching 'cmd_list' """ clsname = self.load_mod(gname,modname) tmpdir_nums = globals()[clsname].tmpdir_nums cdata = [] for a,b in getattr(globals()[clsname],'cmd_group'): if is3seed: for n,(i,j) in enumerate(zip(tmpdir_nums,(128,192,256))): k = '{}_{}'.format(a,n+1) if type(b) == str: cdata.append( (k, (i,'{} ({}-bit)'.format(b,j),[[[],i]])) ) else: cdata.append( (k, (i,'{} ({}-bit)'.format(b[1],j),[[b[0],i]])) ) else: cdata.append( (a, b if full_data else (tmpdir_nums[0],b,[[[],tmpdir_nums[0]]])) ) if add_dpy: self.dpy_data.update(dict(cdata)) else: self.cmd_list = tuple(e[0] for e in cdata) self.dpy_data = dict(cdata) return clsname def init_group(self,trunner,gname,spawn_prog): clsname,kwargs = self.cmd_groups[gname] self.create_group(gname,**kwargs) return globals()[clsname](trunner,cfgs,spawn_prog) def find_cmd_in_groups(self,cmd,group=None): """ Search for a test command in specified group or all configured command groups and return it as a string. Loads modules but alters no global variables. """ if group: if not group in [e[0] for e in self.cmd_groups]: die(1,'{!r}: unrecognized group'.format(group)) groups = [self.cmd_groups[group]] else: groups = self.cmd_groups for gname in groups: clsname,kwargs = self.cmd_groups[gname] self.load_mod(gname,kwargs['modname'] if 'modname' in kwargs else None) if cmd in dict(globals()[clsname].cmd_group): # first search the class return gname if cmd in dir(globals()[clsname](None,None,None)): # then a throwaway instance return gname # cmd might be in several groups - we'll go with the first return None class TestSuiteRunner(object): 'test suite runner' def __init__(self,data_dir,trash_dir): self.data_dir = data_dir self.trash_dir = trash_dir self.cmd_total = 0 from collections import OrderedDict self.rebuild_list = OrderedDict() self.gm = CmdGroupMgr() if opt.log: self.log_fd = open(log_file,'a') self.log_fd.write('\nLog started: {} UTC\n'.format(make_timestr())) omsg('INFO → Logging to file {!r}'.format(log_file)) else: self.log_fd = None if opt.coverage: self.coverdir,self.accfile = init_coverage() omsg('INFO → Writing coverage files to {!r}'.format(self.coverdir)) def spawn_wrapper( self, cmd, args = [], extra_desc = '', no_output = False, msg_only = False, no_msg = False, cmd_dir = 'cmds' ): desc = self.ts.test_name if opt.names else self.gm.dpy_data[self.ts.test_name][1] if extra_desc: desc += ' ' + extra_desc if not opt.system: cmd = os.path.relpath(os.path.join(repo_root,cmd_dir,cmd)) elif g.platform == 'win': cmd = os.path.join('/mingw64','opt','bin',cmd) passthru_opts = ['--{}{}'.format(k.replace('_','-'), '=' + getattr(opt,k) if getattr(opt,k) != True else '') for k in self.ts.passthru_opts if getattr(opt,k)] args = [cmd] + passthru_opts + ['--data-dir='+self.data_dir] + args if opt.traceback: args = ['scripts/traceback_run.py'] + args if g.platform == 'win': args = ['python3'] + args for i in args: if type(i) != str: m = 'Error: missing input files in cmd line?:\nName: {}\nCmdline: {!r}' die(2,m.format(self.ts.test_name,args)) if opt.coverage: args = ['python3','-m','trace','--count','--coverdir='+self.coverdir,'--file='+self.accfile] + args qargs = ['{q}{}{q}'.format(a,q=('',"'")[' ' in a]) for a in args] cmd_disp = ' '.join(qargs).replace('\\','/') # for mingw if not no_msg: if opt.verbose or opt.print_cmdline or opt.exact_output: clr1,clr2 = ((green,cyan),(nocolor,nocolor))[bool(opt.print_cmdline)] omsg(green('Testing: {}'.format(desc))) if not msg_only: s = repr(cmd_disp) if g.platform == 'win' else cmd_disp omsg(clr1('Executing: ') + clr2(s)) else: omsg_r('Testing {}: '.format(desc)) if msg_only: return if opt.log: try: self.log_fd.write(cmd_disp+'\n') except: self.log_fd.write(ascii(cmd_disp)+'\n') from test.pexpect import MMGenPexpect return MMGenPexpect(args,no_output=no_output) def end_msg(self): t = int(time.time()) - self.start_time m = '{} test{} performed. Elapsed time: {:02d}:{:02d}\n' sys.stderr.write(green(m.format(self.cmd_total,suf(self.cmd_total),t//60,t%60))) def init_group(self,gname,cmd=None): ts_cls = globals()[CmdGroupMgr().load_mod(gname)] for k in ('segwit','segwit_random','bech32'): if getattr(opt,k): segwit_opt = k break else: segwit_opt = None m1 = ('test group {g!r}','{g}:{c}')[bool(cmd)].format(g=gname,c=cmd) m2 = ' for {} {}net'.format(g.coin.lower(),'test' if g.testnet else 'main') \ if len(ts_cls.networks) != 1 else '' m3 = ' (--{})'.format(segwit_opt.replace('_','-')) if segwit_opt else '' m = m1 + m2 + m3 if segwit_opt and not getattr(ts_cls,'segwit_opts_ok'): iqmsg('INFO → skipping ' + m) return False # 'networks = ()' means all networks allowed nws = [(e.split('_')[0],'testnet') if '_' in e else (e,'mainnet') for e in ts_cls.networks] if nws: coin = g.coin.lower() nw = ('mainnet','testnet')[g.testnet] for a,b in nws: if a == coin and b == nw: break else: iqmsg('INFO → skipping ' + m) return False bmsg('Executing ' + m) self.ts = self.gm.init_group(self,gname,self.spawn_wrapper) if opt.exit_after and opt.exit_after not in self.gm.cmd_list: die(1,'{!r}: command not recognized'.format(opt.exit_after)) return True def run_tests(self,usr_args): self.start_time = int(time.time()) if usr_args: for arg in usr_args: if arg in self.gm.cmd_groups: if not self.init_group(arg): continue clean(self.ts.tmpdir_nums) for cmd in self.gm.cmd_list: self.check_needs_rerun(cmd,build=True) do_between() elif arg in utils: params = usr_args[usr_args.index(arg)+1:] globals()[arg](*params) sys.exit(0) else: if ':' in arg: gname,arg = arg.split(':') else: gname = self.gm.find_cmd_in_groups(arg) if gname: if not self.init_group(gname,arg): continue clean(self.ts.tmpdir_nums) self.check_needs_rerun(arg,build=True) do_between() else: die(1,'{!r}: command not recognized'.format(arg)) else: if opt.exclude_groups: exclude = opt.exclude_groups.split(',') for e in exclude: if e not in self.gm.dfl_groups: die(1,'{!r}: group not recognized'.format(e)) for gname in self.gm.dfl_groups: if opt.exclude_groups and gname in exclude: continue if not self.init_group(gname): continue clean(self.ts.tmpdir_nums) for cmd in self.gm.cmd_list: self.check_needs_rerun(cmd,build=True) do_between() self.end_msg() def check_needs_rerun(self, cmd, build=False, root=True, force_delete=False, dpy=False ): rerun = root # force_delete is not passed to recursive call fns = [] if force_delete or not root: # does cmd produce a needed dependency(ies)? ret = self.get_num_exts_for_cmd(cmd,dpy) if ret: for ext in ret[1]: fn = get_file_with_ext(cfgs[ret[0]]['tmpdir'],ext,delete=build) if fn: if force_delete: os.unlink(fn) else: fns.append(fn) else: rerun = True fdeps = self.generate_file_deps(cmd) cdeps = self.generate_cmd_deps(fdeps) for fn in fns: my_age = os.stat(fn).st_mtime for num,ext in fdeps: f = get_file_with_ext(cfgs[num]['tmpdir'],ext,delete=build) if f and os.stat(f).st_mtime > my_age: rerun = True for cdep in cdeps: if self.check_needs_rerun(cdep,build=build,root=False,dpy=cmd): rerun = True if build: if rerun: for fn in fns: if not root: os.unlink(fn) if not (dpy and opt.skip_deps): self.run_test(cmd) if not root: do_between() else: # If prog produces multiple files: if cmd not in self.rebuild_list or rerun == True: self.rebuild_list[cmd] = (rerun,fns[0] if fns else '') # FIX return rerun def run_test(self,cmd): # delete files produced by this cmd # for ext,tmpdir in find_generated_exts(cmd): # print cmd, get_file_with_ext(tmpdir,ext) d = [(str(num),ext) for exts,num in self.gm.dpy_data[cmd][2] for ext in exts] # delete files depended on by this cmd arg_list = [get_file_with_ext(cfgs[num]['tmpdir'],ext) for num,ext in d] if opt.resume: if cmd == opt.resume: bmsg('Resuming at {!r}'.format(cmd)) opt.resume = False opt.skip_deps = False else: return if opt.profile: start = time.time() cdata = self.gm.dpy_data[cmd] self.ts.test_name = cmd # self.ts.test_dpydata = cdata self.ts.tmpdir_num = cdata[0] # self.ts.cfg = cfgs[str(cdata[0])] # will remove this eventually cfg = cfgs[str(cdata[0])] for k in ( 'seed_len', 'seed_id', 'wpasswd', 'kapasswd', 'segwit', 'hash_preset', 'bw_filename', 'bw_params', 'ref_bw_seed_id', 'addr_idx_list', 'pass_idx_list' ): if k in cfg: setattr(self.ts,k,cfg[k]) self.process_retval(cmd,getattr(self.ts,cmd)(*arg_list)) # run the test if opt.profile: omsg('\r\033[50C{:.4f}'.format(time.time() - start)) if cmd == opt.exit_after: sys.exit(0) def process_retval(self,cmd,ret): if type(ret).__name__ == 'MMGenPexpect': ret.ok() self.cmd_total += 1 elif ret == 'ok': ok() self.cmd_total += 1 elif ret == 'skip': pass else: rdie(1,'{!r} returned {}'.format(cmd,ret)) def check_deps(self,cmds): # TODO: broken if len(cmds) != 1: die(1,'Usage: {} check_deps '.format(g.prog_name)) cmd = cmds[0] if cmd not in self.gm.cmd_list: die(1,'{!r}: unrecognized command'.format(cmd)) if not opt.quiet: omsg('Checking dependencies for {!r}'.format(cmd)) self.check_needs_rerun(self.ts,cmd,build=False) w = max(map(len,self.rebuild_list)) + 1 for cmd in self.rebuild_list: c = self.rebuild_list[cmd] m = 'Rebuild' if (c[0] and c[1]) else 'Build' if c[0] else 'OK' omsg('cmd {:<{w}} {}'.format(cmd+':', m, w=w)) def generate_file_deps(self,cmd): return [(str(n),e) for exts,n in self.gm.dpy_data[cmd][2] for e in exts] def generate_cmd_deps(self,fdeps): return [cfgs[str(n)]['dep_generators'][ext] for n,ext in fdeps] def get_num_exts_for_cmd(self,cmd,dpy=False): # dpy ignored here try: num = str(self.gm.dpy_data[cmd][0]) except KeyError: qmsg_r('Missing dependency {!r}'.format(cmd)) gname = self.gm.find_cmd_in_groups(cmd) if gname: kwargs = self.gm.cmd_groups[gname][1] kwargs.update({'add_dpy':True}) self.gm.create_group(gname,**kwargs) num = str(self.gm.dpy_data[cmd][0]) qmsg(' found in group {!r}'.format(gname)) else: qmsg(' not found in any command group!') raise dgl = cfgs[num]['dep_generators'] if cmd in dgl.values(): exts = [k for k in dgl if dgl[k] == cmd] return (num,exts) else: return None # main() if not opt.skip_deps: # do this before list cmds exit, so we stay in sync with shm_dir create_tmp_dirs(shm_dir) if opt.list_cmd_groups: Die(0,' '.join(CmdGroupMgr.cmd_groups)) elif opt.list_cmds: list_cmds() if opt.pause: set_restore_term_at_exit() set_environ_for_spawned_scripts() try: tr = TestSuiteRunner(data_dir,trash_dir) tr.run_tests(usr_args) except KeyboardInterrupt: die(1,'\nExiting at user request') except TestSuiteException as e: ydie(1,e.args[0]) except TestSuiteFatalException as e: rdie(1,e.args[0]) except Exception: if opt.traceback: import traceback print(''.join(traceback.format_exception(*sys.exc_info()))) try: os.stat('my.err') t = open('my.err').readlines() if t: msg_r('\n'+yellow(''.join(t[:-1]))+red(t[-1])) except: pass die(1,blue('Test script exited with error')) else: raise except: raise