new class: CmdTestAutosignThreaded; relocate methods

test/cmdtest_py_d/ct_autosign.py

@@ -20,12 +20,12 @@
 test.cmdtest_py_d.ct_autosign: Autosign tests for the cmdtest.py test suite
 """
 
-import sys,os,shutil
+import sys,os,time,shutil
 from subprocess import run,DEVNULL
 from pathlib import Path
 
 from mmgen.cfg import Config
-from mmgen.color import red,green,blue,purple
+from mmgen.color import red,green,blue,yellow,purple,gray
 from mmgen.util import msg,suf,die
 from mmgen.led import LEDControl
 from mmgen.autosign import Autosign
@@ -34,6 +34,8 @@ from ..include.common import (
 	cfg,
 	omsg,
 	omsg_r,
+	oqmsg,
+	oqmsg_r,
 	start_test_daemons,
 	stop_test_daemons,
 	joinpath,
@@ -45,32 +47,12 @@ from .common import ref_dir,dfl_words_file,dfl_bip39_file
 from .ct_base import CmdTestBase
 from .input import stealth_mnemonic_entry
 
-filedir_map = (
-	('btc',''),
-	('bch',''),
-	('ltc','litecoin'),
-	('eth','ethereum'),
-	('mm1','ethereum'),
-	('etc','ethereum_classic'),
-)
-
-def init_led(simulate):
-	try:
-		cf = LEDControl(enabled=True,simulate=simulate)
-	except Exception as e:
-		msg(str(e))
-		die(2,'LEDControl initialization failed')
-	for fn in (cf.board.status,cf.board.trigger):
-		if fn:
-			run(['sudo','chmod','0666',fn],check=True)
-
 class CmdTestAutosignBase(CmdTestBase):
 	networks     = ('btc',)
 	tmpdir_nums  = [18]
 	color        = True
 	win_skip     = True
-	no_insert_check = True
-	have_online = False
+	daemon_coins = []
 
 	def __init__(self,trunner,cfgs,spawn):
 
@@ -87,36 +69,8 @@ class CmdTestAutosignBase(CmdTestBase):
 		if not (cfg.skipping_deps or self.live):
 			self._create_removable_device()
 
-		if self.simulate_led and not cfg.exact_output:
-			die(1,red('This command must be run with --exact-output enabled!'))
-
-		if self.simulate_led or not self.live:
-			LEDControl.create_dummy_control_files()
-			self.spawn_env['MMGEN_TEST_SUITE_AUTOSIGN_LED_SIMULATE'] = '1'
-
 		self.opts = ['--coins='+','.join(self.coins)]
 
-		if self.live:
-			init_led(self.simulate_led)
-		elif self.no_insert_check:
-			self.opts.append('--no-insert-check')
-
-		self.tx_file_ops('set_count') # initialize tx_count here so we can resume anywhere
-
-		def gen_msg_fns():
-			fmap = dict(filedir_map)
-			for coin in self.coins:
-				if coin == 'xmr':
-					continue
-				sdir = os.path.join('test','ref',fmap[coin])
-				for fn in os.listdir(sdir):
-					if fn.endswith(f'[{coin.upper()}].rawmsg.json'):
-						yield os.path.join(sdir,fn)
-
-		self.ref_msgfiles = tuple(gen_msg_fns())
-		self.good_msg_count = 0
-		self.bad_msg_count = 0
-
 		if not self.live:
 			self.spawn_env['MMGEN_TEST_SUITE_ROOT_PFX'] = self.tmpdir
 
@@ -150,12 +104,6 @@ class CmdTestAutosignBase(CmdTestBase):
 		(self.asi.mountpoint / 'tx').mkdir()
 		self.do_umount()
 
-	def __del__(self):
-		if sys.platform == 'win32' or self.tr is None:
-			return
-		if self.simulate_led or not self.live:
-			LEDControl.delete_dummy_control_files()
-
 	def start_daemons(self):
 		self.spawn('',msg_only=True)
 		start_test_daemons(*self.network_ids)
@@ -166,28 +114,6 @@ class CmdTestAutosignBase(CmdTestBase):
 		stop_test_daemons(*self.network_ids)
 		return 'ok'
 
-	def gen_key(self):
-		t = self.spawn( 'mmgen-autosign', self.opts + ['gen_key'] )
-		t.expect_getend('Wrote key file ')
-		return t
-
-	def create_dfl_wallet(self):
-		t = self.spawn( 'mmgen-walletconv', [
-				f'--outdir={cfg.data_dir}',
-				'--usr-randchars=0', '--quiet', '--hash-preset=1', '--label=foo',
-				'test/ref/98831F3A.hex'
-			]
-		)
-		t.passphrase_new('new MMGen wallet','abc')
-		t.written_to_file('MMGen wallet')
-		return t
-
-	def run_setup_dfl_wallet(self):
-		return self.run_setup(mn_type='default',use_dfl_wallet=True)
-
-	def run_setup_bip39(self):
-		return self.run_setup(mn_type='bip39')
-
 	def run_setup(self,mn_type=None,mn_file=None,use_dfl_wallet=False):
 		mn_desc = mn_type or 'default'
 		mn_type = mn_type or 'mmgen'
@@ -218,6 +144,198 @@ class CmdTestAutosignBase(CmdTestBase):
 		t.written_to_file('Autosign wallet')
 		return t
 
+	def _mount_ops(self, loc, cmd, *args, **kwargs):
+		return getattr(getattr(self,loc),cmd)(*args, silent=self.silent_mount, **kwargs)
+
+	def do_mount(self, *args, **kwargs):
+		return self._mount_ops('asi', 'do_mount', *args, **kwargs)
+
+	def do_umount(self, *args, **kwargs):
+		return self._mount_ops('asi', 'do_umount', *args, **kwargs)
+
+	def do_mount_online(self, *args, **kwargs):
+		return self._mount_ops('asi_online', 'do_mount', *args, **kwargs)
+
+	def do_umount_online(self, *args, **kwargs):
+		return self._mount_ops('asi_online', 'do_umount', *args, **kwargs)
+
+class CmdTestAutosignThreaded(CmdTestAutosignBase):
+	have_online     = True
+	live            = False
+	no_insert_check = False
+
+	def autosign_start_thread(self):
+		def run():
+			t = self.spawn('mmgen-autosign', self.opts + ['wait'], direct_exec=True)
+			self.write_to_tmpfile('autosign_thread_pid',str(t.ep.pid))
+		import threading
+		threading.Thread(target=run, name='Autosign wait loop').start()
+		time.sleep(0.2)
+		return 'silent'
+
+	def autosign_kill_thread(self):
+		self.spawn('',msg_only=True)
+		pid = int(self.read_from_tmpfile('autosign_thread_pid'))
+		self.delete_tmpfile('autosign_thread_pid')
+		from signal import SIGTERM
+		imsg(purple(f'Killing autosign wait loop [PID {pid}]'))
+		try:
+			os.kill(pid,SIGTERM)
+		except:
+			imsg(yellow(f'{pid}: no such process'))
+		return 'ok'
+
+	def _wait_signed(self,desc):
+		oqmsg_r(gray(f'→ offline wallet{"s" if desc.endswith("s") else ""} signing {desc}'))
+		assert not self.device_inserted, f'‘{self.asi.dev_label_path}’ is inserted!'
+		assert not self.asi.mountpoint.is_mount(), f'‘{self.asi.mountpoint}’ is mounted!'
+		self.insert_device()
+		while True:
+			oqmsg_r(gray('.'))
+			if self.asi.mountpoint.is_mount():
+				oqmsg_r(gray('..working..'))
+				break
+			time.sleep(0.5)
+		while True:
+			oqmsg_r(gray('.'))
+			if not self.asi.mountpoint.is_mount():
+				oqmsg(gray('..done'))
+				break
+			time.sleep(0.5)
+		self.remove_device()
+
+	@property
+	def device_inserted(self):
+		return self.asi.dev_label_path.exists()
+
+	def insert_device(self):
+		self.asi.dev_label_path.touch()
+
+	def remove_device(self):
+		if self.asi.dev_label_path.exists():
+			self.asi.dev_label_path.unlink()
+
+	@property
+	def device_inserted_online(self):
+		return self.asi_online.dev_label_path.exists()
+
+	def insert_device_online(self):
+		self.asi_online.dev_label_path.touch()
+
+	def remove_device_online(self):
+		if self.asi_online.dev_label_path.exists():
+			self.asi_online.dev_label_path.unlink()
+
+class CmdTestAutosign(CmdTestAutosignBase):
+	'autosigning transactions for all supported coins'
+	coins           = ['btc','bch','ltc','eth']
+	daemon_coins    = ['btc','bch','ltc']
+	txfile_coins    = ['btc','bch','ltc','eth','mm1','etc']
+	have_online     = False
+	live            = False
+	simulate_led    = True
+	no_insert_check = True
+
+	filedir_map = (
+		('btc',''),
+		('bch',''),
+		('ltc','litecoin'),
+		('eth','ethereum'),
+		('mm1','ethereum'),
+		('etc','ethereum_classic'),
+	)
+
+	cmd_group = (
+		('start_daemons',            'starting daemons'),
+		('copy_tx_files',            'copying transaction files'),
+		('gen_key',                  'generating key'),
+		('create_dfl_wallet',        'creating default MMGen wallet'),
+		('run_setup_dfl_wallet',     'running ‘autosign setup’ (with default wallet)'),
+		('sign_quiet',               'signing transactions (--quiet)'),
+		('remove_signed_txfiles',    'removing signed transaction files'),
+		('run_setup_bip39',          'running ‘autosign setup’ (BIP39 mnemonic)'),
+		('create_bad_txfiles',       'creating bad transaction files'),
+		('sign_full_summary',        'signing transactions (--full-summary)'),
+		('remove_signed_txfiles_btc','removing transaction files (BTC only)'),
+		('remove_bad_txfiles',       'removing bad transaction files'),
+		('sign_led',                 'signing transactions (--led - BTC files only)'),
+		('remove_signed_txfiles',    'removing signed transaction files'),
+		('sign_stealth_led',         'signing transactions (--stealth-led)'),
+		('remove_signed_txfiles',    'removing signed transaction files'),
+		('copy_msgfiles',            'copying message files'),
+		('sign_quiet_msg',           'signing transactions and messages (--quiet)'),
+		('remove_signed_txfiles',    'removing signed transaction files'),
+		('create_bad_txfiles2',      'creating bad transaction files'),
+		('remove_signed_msgfiles',   'removing signed message files'),
+		('create_invalid_msgfile',   'creating invalid message file'),
+		('sign_full_summary_msg',    'signing transactions and messages (--full-summary)'),
+		('remove_invalid_msgfile',   'removing invalid message file'),
+		('remove_bad_txfiles2',      'removing bad transaction files'),
+		('sign_no_unsigned_msg',     'signing transactions and messages (nothing to sign)'),
+		('stop_daemons',             'stopping daemons'),
+	)
+
+	def __init__(self,trunner,cfgs,spawn):
+
+		super().__init__(trunner,cfgs,spawn)
+
+		if trunner is None:
+			return
+
+		if self.live and not cfg.exact_output:
+			die(1,red('autosign_live tests must be run with --exact-output enabled!'))
+
+		if self.no_insert_check:
+			self.opts.append('--no-insert-check')
+
+		self.tx_file_ops('set_count') # initialize self.tx_count here so we can resume anywhere
+		self.bad_tx_count = 0
+
+		def gen_msg_fns():
+			fmap = dict(self.filedir_map)
+			for coin in self.coins:
+				if coin == 'xmr':
+					continue
+				sdir = os.path.join('test','ref',fmap[coin])
+				for fn in os.listdir(sdir):
+					if fn.endswith(f'[{coin.upper()}].rawmsg.json'):
+						yield os.path.join(sdir,fn)
+
+		self.ref_msgfiles = tuple(gen_msg_fns())
+		self.good_msg_count = 0
+		self.bad_msg_count = 0
+
+		if self.simulate_led:
+			LEDControl.create_dummy_control_files()
+			self.have_dummy_control_files = True
+			self.spawn_env['MMGEN_TEST_SUITE_AUTOSIGN_LED_SIMULATE'] = '1'
+
+	def __del__(self):
+		if hasattr(self,'have_dummy_control_files'):
+			LEDControl.delete_dummy_control_files()
+
+	def gen_key(self):
+		t = self.spawn( 'mmgen-autosign', self.opts + ['gen_key'] )
+		t.expect_getend('Wrote key file ')
+		return t
+
+	def create_dfl_wallet(self):
+		t = self.spawn( 'mmgen-walletconv', [
+				f'--outdir={cfg.data_dir}',
+				'--usr-randchars=0', '--quiet', '--hash-preset=1', '--label=foo',
+				'test/ref/98831F3A.hex'
+			]
+		)
+		t.passphrase_new('new MMGen wallet','abc')
+		t.written_to_file('MMGen wallet')
+		return t
+
+	def run_setup_dfl_wallet(self):
+		return self.run_setup(mn_type='default',use_dfl_wallet=True)
+
+	def run_setup_bip39(self):
+		return self.run_setup(mn_type='bip39')
+
 	def copy_tx_files(self):
 		self.spawn('',msg_only=True)
 		return self.tx_file_ops('copy')
@@ -237,7 +355,7 @@ class CmdTestAutosignBase(CmdTestBase):
 		from .ct_ref import CmdTestRef
 		def gen():
 			d = CmdTestRef.sources['ref_tx_file']
-			dirmap = [e for e in filedir_map if e[0] in (txfile_coins or self.txfile_coins)]
+			dirmap = [e for e in self.filedir_map if e[0] in (txfile_coins or self.txfile_coins)]
 			for coin,coindir in dirmap:
 				for network in (0,1):
 					fn = d[coin][network]
@@ -371,81 +489,6 @@ class CmdTestAutosignBase(CmdTestBase):
 		imsg('')
 		return t
 
-	@property
-	def device_inserted(self):
-		return self.asi.dev_label_path.exists()
-
-	@property
-	def device_inserted_online(self):
-		return self.asi_online.dev_label_path.exists()
-
-	def insert_device(self):
-		self.asi.dev_label_path.touch()
-
-	def insert_device_online(self):
-		self.asi_online.dev_label_path.touch()
-
-	def remove_device(self):
-		if self.asi.dev_label_path.exists():
-			self.asi.dev_label_path.unlink()
-
-	def remove_device_online(self):
-		if self.asi_online.dev_label_path.exists():
-			self.asi_online.dev_label_path.unlink()
-
-	def _mount_ops(self, loc, cmd, *args, **kwargs):
-		return getattr(getattr(self,loc),cmd)(*args, silent=self.silent_mount, **kwargs)
-
-	def do_mount(self, *args, **kwargs):
-		return self._mount_ops('asi', 'do_mount', *args, **kwargs)
-
-	def do_umount(self, *args, **kwargs):
-		return self._mount_ops('asi', 'do_umount', *args, **kwargs)
-
-	def do_mount_online(self, *args, **kwargs):
-		return self._mount_ops('asi_online', 'do_mount', *args, **kwargs)
-
-	def do_umount_online(self, *args, **kwargs):
-		return self._mount_ops('asi_online', 'do_umount', *args, **kwargs)
-
-class CmdTestAutosign(CmdTestAutosignBase):
-	'autosigning transactions for all supported coins'
-	coins        = ['btc','bch','ltc','eth']
-	daemon_coins = ['btc','bch','ltc']
-	txfile_coins = ['btc','bch','ltc','eth','mm1','etc']
-	live         = False
-	simulate_led = False
-	bad_tx_count = 0
-	cmd_group = (
-		('start_daemons',            'starting daemons'),
-		('copy_tx_files',            'copying transaction files'),
-		('gen_key',                  'generating key'),
-		('create_dfl_wallet',        'creating default MMGen wallet'),
-		('run_setup_dfl_wallet',     'running ‘autosign setup’ (with default wallet)'),
-		('sign_quiet',               'signing transactions (--quiet)'),
-		('remove_signed_txfiles',    'removing signed transaction files'),
-		('run_setup_bip39',          'running ‘autosign setup’ (BIP39 mnemonic)'),
-		('create_bad_txfiles',       'creating bad transaction files'),
-		('sign_full_summary',        'signing transactions (--full-summary)'),
-		('remove_signed_txfiles_btc','removing transaction files (BTC only)'),
-		('remove_bad_txfiles',       'removing bad transaction files'),
-		('sign_led',                 'signing transactions (--led - BTC files only)'),
-		('remove_signed_txfiles',    'removing signed transaction files'),
-		('sign_stealth_led',         'signing transactions (--stealth-led)'),
-		('remove_signed_txfiles',    'removing signed transaction files'),
-		('copy_msgfiles',            'copying message files'),
-		('sign_quiet_msg',           'signing transactions and messages (--quiet)'),
-		('remove_signed_txfiles',    'removing signed transaction files'),
-		('create_bad_txfiles2',      'creating bad transaction files'),
-		('remove_signed_msgfiles',   'removing signed message files'),
-		('create_invalid_msgfile',   'creating invalid message file'),
-		('sign_full_summary_msg',    'signing transactions and messages (--full-summary)'),
-		('remove_invalid_msgfile',   'removing invalid message file'),
-		('remove_bad_txfiles2',      'removing bad transaction files'),
-		('sign_no_unsigned_msg',     'signing transactions and messages (nothing to sign)'),
-		('stop_daemons',             'stopping daemons'),
-	)
-
 	def sign_quiet(self):
 		return self.do_sign(['--quiet','wait'])
 
@@ -478,7 +521,10 @@ class CmdTestAutosignBTC(CmdTestAutosign):
 
 class CmdTestAutosignLive(CmdTestAutosignBTC):
 	'live autosigning BTC transactions'
-	live = True
+	live            = True
+	simulate_led    = False
+	no_insert_check = False
+
 	cmd_group = (
 		('start_daemons',        'starting daemons'),
 		('copy_tx_files',        'copying transaction files'),
@@ -492,6 +538,22 @@ class CmdTestAutosignLive(CmdTestAutosignBTC):
 		('stop_daemons',         'stopping daemons'),
 	)
 
+	def __init__(self,trunner,cfgs,spawn):
+
+		super().__init__(trunner,cfgs,spawn)
+
+		if trunner is None:
+			return
+
+		try:
+			cf = LEDControl(enabled=True,simulate=self.simulate_led)
+		except Exception as e:
+			msg(str(e))
+			die(2,'LEDControl initialization failed')
+		for path in (cf.board.status,cf.board.trigger):
+			if path:
+				run(['sudo','chmod','0666',path],check=True)
+
 	def run_setup_mmgen(self):
 		return self.run_setup(mn_type='mmgen',use_dfl_wallet=None)
 

test/cmdtest_py_d/ct_xmr_autosign.py

@@ -16,21 +16,14 @@ test.cmdtest_py_d.ct_xmr_autosign: xmr autosigning tests for the cmdtest.py test
 import os,time,re,shutil
 from pathlib import Path
 
-from mmgen.color import yellow,purple,gray
+from mmgen.color import purple,gray
 from mmgen.util import fmt,async_run
 
-from ..include.common import (
-	cfg,
-	oqmsg,
-	oqmsg_r,
-	imsg,
-	silence,
-	end_silence
-)
+from ..include.common import cfg,imsg,silence,end_silence
 from .common import get_file_with_ext
 
 from .ct_xmrwallet import CmdTestXMRWallet
-from .ct_autosign import CmdTestAutosignBase
+from .ct_autosign import CmdTestAutosignThreaded
 
 def make_burn_addr():
 	from mmgen.tool.coin import tool_cmd
@@ -40,7 +33,7 @@ def make_burn_addr():
 		proto   = cfg._proto,
 		mmtype  = 'monero' ).privhex2addr('beadcafe'*8)
 
-class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
+class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignThreaded):
 	"""
 	Monero autosigning operations
 	"""
@@ -54,15 +47,7 @@ class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
 	)
 
 	# ct_autosign attrs:
-	coins        = ['xmr']
-	daemon_coins = []
-	txfile_coins = []
-	live         = False
-	simulate_led = False
-	bad_tx_count = 0
-	no_insert_check = False
-	win_skip = True
-	have_online = True
+	coins = ['xmr']
 
 	cmd_group = (
 		('daemon_version',           'checking daemon version'),
@@ -108,7 +93,7 @@ class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
 
 	def __init__(self,trunner,cfgs,spawn):
 
-		CmdTestAutosignBase.__init__(self,trunner,cfgs,spawn)
+		CmdTestAutosignThreaded.__init__(self,trunner,cfgs,spawn)
 		CmdTestXMRWallet.__init__(self,trunner,cfgs,spawn)
 
 		if trunner is None:
@@ -127,7 +112,6 @@ class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
 
 		self.opts.append('--xmrwallets={}'.format(self.users['alice'].kal_range)) # mmgen-autosign opts
 		self.autosign_opts = ['--autosign']                                       # mmgen-xmrwallet opts
-		self.tx_count = 1
 		self.spawn_env['MMGEN_TEST_SUITE_XMR_AUTOSIGN'] = '1'
 
 	def create_tmp_wallets(self):
@@ -245,27 +229,6 @@ class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
 
 		return t
 
-	def autosign_start_thread(self):
-		def run():
-			t = self.spawn('mmgen-autosign', self.opts + ['wait'], direct_exec=True)
-			self.write_to_tmpfile('autosign_thread_pid',str(t.ep.pid))
-		import threading
-		threading.Thread( target=run, name='Autosign wait loop' ).start()
-		time.sleep(0.2)
-		return 'silent'
-
-	def autosign_kill_thread(self):
-		self.spawn('',msg_only=True)
-		pid = int(self.read_from_tmpfile('autosign_thread_pid'))
-		self.delete_tmpfile('autosign_thread_pid')
-		from signal import SIGTERM
-		imsg(purple(f'Killing autosign wait loop [PID {pid}]'))
-		try:
-			os.kill(pid,SIGTERM)
-		except:
-			imsg(yellow(f'{pid}: no such process'))
-		return 'ok'
-
 	def create_watchonly_wallets(self):
 		self.insert_device_online()
 		t = self.create_wallets('alice', op='restore')
@@ -293,25 +256,6 @@ class CmdTestXMRAutosign(CmdTestXMRWallet,CmdTestAutosignBase):
 		self.do_umount_online()
 		return self._create_transfer_tx('0.257')
 
-	def _wait_signed(self,desc):
-		oqmsg_r(gray(f'→ offline wallet{"s" if desc.endswith("s") else ""} signing {desc}'))
-		assert not self.device_inserted, f'‘{self.asi.dev_label_path}’ is inserted!'
-		assert not self.asi.mountpoint.is_mount(), f'‘{self.asi.mountpoint}’ is mounted!'
-		self.insert_device()
-		while True:
-			oqmsg_r(gray('.'))
-			if self.asi.mountpoint.is_mount():
-				oqmsg_r(gray('..working..'))
-				break
-			time.sleep(0.5)
-		while True:
-			oqmsg_r(gray('.'))
-			if not self.asi.mountpoint.is_mount():
-				oqmsg(gray('..done'))
-				break
-			time.sleep(0.5)
-		self.remove_device()
-
 	def _xmr_autosign_op(
 			self,
 			op,