From e9b0c9e325f093d2d0e2b8b90e51c86c8475d632 Mon Sep 17 00:00:00 2001 From: The MMGen Project Date: Sun, 26 Sep 2021 21:16:19 +0000 Subject: [PATCH] unit_tests.py daemon: refactor to use subtest API --- test/unit_tests_d/ut_daemon.py | 152 +++++++++++++++++++-------------- 1 file changed, 86 insertions(+), 66 deletions(-) diff --git a/test/unit_tests_d/ut_daemon.py b/test/unit_tests_d/ut_daemon.py index 15c6ce85..d8e9c8ce 100755 --- a/test/unit_tests_d/ut_daemon.py +++ b/test/unit_tests_d/ut_daemon.py @@ -9,75 +9,82 @@ from mmgen.exception import * from mmgen.daemon import * from mmgen.protocol import init_proto -class unit_test(object): +def test_flags(): + d = CoinDaemon('eth') + vmsg(f'Available opts: {fmt_list(d.avail_opts,fmt="bare")}') + vmsg(f'Available flags: {fmt_list(d.avail_flags,fmt="bare")}') + vals = namedtuple('vals',['online','no_daemonize','keep_cfg_file']) - def run_test(self,name,ut): + def gen(): + for opts,flags,val in ( + (None,None, vals(False,False,False)), + (None,['keep_cfg_file'], vals(False,False,True)), + (['online'],['keep_cfg_file'], vals(True,False,True)), + (['online','no_daemonize'],['keep_cfg_file'], vals(True,True,True)), + ): + d = CoinDaemon('eth',opts=opts,flags=flags) + assert d.flag.keep_cfg_file == val.keep_cfg_file + assert d.opt.online == val.online + assert d.opt.no_daemonize == val.no_daemonize + d.flag.keep_cfg_file = not val.keep_cfg_file + d.flag.keep_cfg_file = val.keep_cfg_file + yield d - def test_flags(): - d = CoinDaemon('eth') - vmsg(f'Available opts: {fmt_list(d.avail_opts,fmt="bare")}') - vmsg(f'Available flags: {fmt_list(d.avail_flags,fmt="bare")}') - vals = namedtuple('vals',['online','no_daemonize','keep_cfg_file']) + return tuple(gen()) - def gen(): - for opts,flags,val in ( - (None,None, vals(False,False,False)), - (None,['keep_cfg_file'], vals(False,False,True)), - (['online'],['keep_cfg_file'], vals(True,False,True)), - (['online','no_daemonize'],['keep_cfg_file'], vals(True,True,True)), - ): - d = CoinDaemon('eth',opts=opts,flags=flags) - assert d.flag.keep_cfg_file == val.keep_cfg_file - assert d.opt.online == val.online - assert d.opt.no_daemonize == val.no_daemonize - d.flag.keep_cfg_file = not val.keep_cfg_file - d.flag.keep_cfg_file = val.keep_cfg_file - yield d +def test_flags_err(ut,d): - return tuple(gen()) + def bad1(): d[0].flag.foo = False + def bad2(): d[0].opt.foo = False + def bad3(): d[0].opt.no_daemonize = True + def bad4(): d[0].flag.keep_cfg_file = 'x' + def bad5(): d[0].opt.no_daemonize = 'x' + def bad6(): d[0].flag.keep_cfg_file = False + def bad7(): d[1].flag.keep_cfg_file = True - def test_flags_err(d): + ut.process_bad_data(( + ('flag (1)', 'ClassFlagsError', 'unrecognized flag', bad1 ), + ('opt (1)', 'ClassFlagsError', 'unrecognized opt', bad2 ), + ('opt (2)', 'AttributeError', 'is read-only', bad3 ), + ('flag (2)', 'AssertionError', 'not boolean', bad4 ), + ('opt (3)', 'AttributeError', 'is read-only', bad5 ), + ('flag (3)', 'ClassFlagsError', 'not set', bad6 ), + ('flag (4)', 'ClassFlagsError', 'already set', bad7 ), + )) - def bad1(): d[0].flag.foo = False - def bad2(): d[0].opt.foo = False - def bad3(): d[0].opt.no_daemonize = True - def bad4(): d[0].flag.keep_cfg_file = 'x' - def bad5(): d[0].opt.no_daemonize = 'x' - def bad6(): d[0].flag.keep_cfg_file = False - def bad7(): d[1].flag.keep_cfg_file = True +def test_cmds(op): + network_ids = CoinDaemon.get_network_ids() + for test_suite in [True,False] if op == 'print' else [True]: + vmsg(orange(f'Start commands (op={op}, test_suite={test_suite}):')) + for coin,data in CoinDaemon.coins.items(): + for daemon_id in data.daemon_ids: + for network in globals()[daemon_id+'_daemon'].networks: + if opt.no_altcoin_deps and coin != 'BTC': + continue + d = CoinDaemon( + proto=init_proto(coin=coin,network=network), + daemon_id = daemon_id, + test_suite = test_suite ) + if op == 'print': + for cmd in d.start_cmds: + vmsg(' '.join(cmd)) + elif op == 'check': + try: + cp = run([d.exec_fn,'--help'],stdout=PIPE,stderr=PIPE) + except: + ydie(1,f'Unable to execute {d.exec_fn}') + if cp.returncode: + ydie(1,f'Unable to execute {d.exec_fn}') + else: + vmsg('{:16} {}'.format(d.exec_fn+':',cp.stdout.decode().splitlines()[0])) + else: + if opt.quiet: + msg_r('.') + getattr(d,op)(silent=opt.quiet) - ut.process_bad_data(( - ('flag (1)', 'ClassFlagsError', 'unrecognized flag', bad1 ), - ('opt (1)', 'ClassFlagsError', 'unrecognized opt', bad2 ), - ('opt (2)', 'AttributeError', 'is read-only', bad3 ), - ('flag (2)', 'AssertionError', 'not boolean', bad4 ), - ('opt (3)', 'AttributeError', 'is read-only', bad5 ), - ('flag (3)', 'ClassFlagsError', 'not set', bad6 ), - ('flag (4)', 'ClassFlagsError', 'already set', bad7 ), - )) +class unit_tests: - def test_cmds(op): - network_ids = CoinDaemon.get_network_ids() - for test_suite in [True,False] if op == 'print' else [True]: - vmsg(orange(f'Start commands (op={op}, test_suite={test_suite}):')) - for coin,data in CoinDaemon.coins.items(): - for daemon_id in data.daemon_ids: - for network in globals()[daemon_id+'_daemon'].networks: - d = CoinDaemon( - proto=init_proto(coin=coin,network=network), - daemon_id = daemon_id, - test_suite = test_suite ) - if op == 'print': - for cmd in d.start_cmds: - vmsg(' '.join(cmd)) - else: - if run(['which',d.exec_fn],stdout=DEVNULL,stderr=DEVNULL).returncode: - if op == 'start': - qmsg(yellow(f'Warning: {d.exec_fn} not found in executable path')) - else: - if opt.quiet: - msg_r('.') - getattr(d,op)(silent=opt.quiet) + def flags(self,name,ut): qmsg_r('Testing flags and opts...') vmsg('') @@ -86,22 +93,35 @@ class unit_test(object): qmsg_r('Testing error handling for flags and opts...') vmsg('') - test_flags_err(daemons) + test_flags_err(ut,daemons) qmsg('OK') - qmsg_r('Testing start commands for configured daemons...') + return True + + def cmds(self,name,ut): + qmsg_r('Testing start commands for coin daemons...') vmsg('') test_cmds('print') qmsg('OK') + return True - msg_r('Starting all configured daemons available on system...') + def exec(self,name,ut): + qmsg_r('Testing availability of coin daemons...') + vmsg('') + test_cmds('check') + qmsg('OK') + return True + + def start(self,name,ut): + msg_r('Starting coin daemons...') qmsg('') test_cmds('start') msg('OK') + return True - msg_r('Stopping all configured daemons available on system...') + def stop(self,name,ut): + msg_r('Stopping coin daemons...') qmsg('') test_cmds('stop') msg('OK') - return True