test.include.common: reimplement `VirtBlockDevice`

mmgen/util.py

@@ -193,6 +193,8 @@ def fmt_list(iterable,fmt='dfl',indent='',conv=None):
 		'dfl':       ( str,  ", ", "'",  "'"),
 		'utf8':      ( str,  ", ", "“",  "”"),
 		'bare':      ( repr, " ",  "",   ""),
+		'barest':    ( str,  " ",  "",   ""),
+		'fancy':     ( str,  " ",  "‘",  "’"),
 		'no_quotes': ( str,  ", ", "",   ""),
 		'compact':   ( str,  ",",  "",   ""),
 		'no_spc':    ( str,  ",",  "'",  "'"),

pyproject.toml

@@ -80,4 +80,5 @@ ignored-classes = [ # ignored for no-member, otherwise checked
 # test suite:
 	"TestHashFunc",
 	"GenTool",
+	"VirtBlockDeviceBase",
 ]

test/cmdtest_py_d/ct_autosign.py

@@ -43,6 +43,7 @@ from ..include.common import (
 	read_from_file,
 	silence,
 	end_silence,
+	VirtBlockDevice,
 )
 from .common import ref_dir,dfl_words_file,dfl_bip39_file
 
@@ -69,6 +70,9 @@ class CmdTestAutosignBase(CmdTestBase):
 
 		self._create_autosign_instances(create_dirs=not cfg.skipping_deps)
 
+		if sys.platform == 'linux':
+			self.txdev = VirtBlockDevice(self.asi.fs_image_path, '10M')
+
 		if not (cfg.skipping_deps or self.live):
 			self._create_removable_device()
 
@@ -86,6 +90,8 @@ class CmdTestAutosignBase(CmdTestBase):
 	def __del__(self):
 		if hasattr(self,'have_dummy_control_files'):
 			LEDControl.delete_dummy_control_files()
+		if hasattr(self, 'txdev'):
+			del self.txdev
 		if not cfg.no_daemon_stop:
 			if sys.platform == 'darwin':
 				for label in (self.asi.dev_label, self.asi.ramdisk.label):
@@ -123,14 +129,16 @@ class CmdTestAutosignBase(CmdTestBase):
 
 	def _create_removable_device(self):
 		if sys.platform == 'linux':
-			run(['truncate', '--size=10M', str(self.asi.fs_image_path)], check=True)
+			self.txdev.create()
+			self.txdev.attach(silent=True)
 			cmd = [
 				'/sbin/mkfs.ext2',
 				'-E', 'root_owner={}:{}'.format(os.getuid(), os.getgid()),
 				'-L', self.asi.dev_label,
-				str(self.asi.fs_image_path)]
+				str(self.txdev.img_path)]
 			redir = DEVNULL
 			run(cmd, stdout=redir, stderr=redir, check=True)
+			self.txdev.detach(silent=True)
 		elif sys.platform == 'darwin':
 			cmd = [
 				'hdiutil', 'create', '-size', '10M', '-fs', 'exFAT',
@@ -227,6 +235,7 @@ class CmdTestAutosignBase(CmdTestBase):
 		loc = getattr(self, asi)
 		if sys.platform == 'linux':
 			loc.dev_label_path.touch()
+			# self.txdev.attach() # WIP
 		elif sys.platform == 'darwin':
 			self._macOS_mount_fs_image(loc)
 
@@ -237,6 +246,7 @@ class CmdTestAutosignBase(CmdTestBase):
 		if sys.platform == 'linux':
 			if loc.dev_label_path.exists():
 				loc.dev_label_path.unlink()
+			# self.txdev.detach() # WIP
 		elif sys.platform == 'darwin':
 			self._macOS_eject_disk(loc.dev_label)
 

test/cmdtest_py_d/ct_wallet.py

@@ -20,7 +20,7 @@
 test.cmdtest_py_d.ct_wallet: Wallet conversion tests for the cmdtest.py test suite
 """
 
-import os
+import sys, os
 
 from mmgen.util import msg,capfirst,get_extension
 from mmgen.wallet import get_wallet_cls
@@ -167,13 +167,14 @@ class CmdTestWalletConv(CmdTestBase,CmdTestShared):
 
 	def ref_hincog_blkdev_conv_out(self):
 
-		if self.skip_for_win('no loop device') or self.skip_for_mac('no loop device'):
+		if self.skip_for_win('no loop device'):
 			return 'skip'
 
-		b = VirtBlockDevice(self.tmpdir, '1K', 1)
-		b.setup()
+		b = VirtBlockDevice(os.path.join(self.tmpdir, 'hincog_blkdev_img'), '1K')
+		b.create()
+		b.attach(dev_mode='0666')
 		self.ref_hincog_conv_out(ic_f=b.dev)
-		b.destroy()
+		b.detach()
 
 		return 'ok'
 

test/include/common.py

@@ -20,8 +20,9 @@
 test.include.common: Shared routines and data for the MMGen test suites
 """
 
-import sys,os
+import sys, os, re, atexit
 from subprocess import run,PIPE
+from pathlib import Path
 
 from mmgen.cfg import gv
 from mmgen.color import yellow,green,orange
@@ -336,27 +337,87 @@ def do_run(cmd, check=True):
 	from subprocess import run,PIPE,DEVNULL
 	return run(cmd, stdout=PIPE, stderr=DEVNULL, check=check)
 
-class VirtBlockDevice:
-
-	def __init__(self, tmpdir, blksize, blkcount):
-		self.tmpdir = tmpdir
-		self.blksize = blksize
-		self.blkcount = blkcount
-
-	def setup(self):
-		imsg('Creating block device image file')
-		blkdev_img = joinpath(self.tmpdir, 'hincog_blkdev_img')
-		do_run(['dd', 'if=/dev/zero', f'of={blkdev_img}', f'bs={self.blksize}', f'count={self.blkcount}'])
-		self.dev = do_run(['sudo', '/sbin/losetup', '-f']).stdout.strip().decode()
-		self.dev_mode_orig = '{:o}'.format(os.stat(self.dev).st_mode & 0xfff)
-		dev_mode = '0666'
-		imsg(f'Changing permissions on loop device to {dev_mode!r}')
-		do_run(['sudo', 'chmod', dev_mode, self.dev])
-		imsg(f'Attaching loop device {self.dev!r}')
-		do_run(['sudo', '/sbin/losetup', self.dev, blkdev_img])
-
-	def destroy(self):
-		imsg(f'Detaching loop device {self.dev!r}')
-		do_run(['sudo', '/sbin/losetup', '-d', self.dev])
-		imsg(f'Resetting permissions on loop device to {self.dev_mode_orig!r}')
-		do_run(['sudo', 'chmod', self.dev_mode_orig, self.dev])
+def VirtBlockDevice(img_path, size):
+	if sys.platform == 'linux':
+		return VirtBlockDeviceLinux(img_path, size)
+	elif sys.platform == 'darwin':
+		return VirtBlockDeviceMacOS(img_path, size)
+
+class VirtBlockDeviceBase:
+
+	@property
+	def dev(self):
+		res = self._get_associations()
+		if len(res) < 1:
+			die(2, f'No device associated with {self.img_path}')
+		elif len(res) > 1:
+			die(2, f'More than one device associated with {self.img_path}')
+		return res[0]
+
+	def try_detach(self):
+		try:
+			dev = self.dev
+		except:
+			pass
+		else:
+			self.do_detach(dev, check=False)
+
+	def create(self, silent=False):
+		for dev in self._get_associations():
+			if not silent:
+				imsg(f'Detaching associated device {dev}')
+			self.do_detach(dev)
+		self.img_path.unlink(missing_ok=True)
+		if not silent:
+			imsg(f'Creating block device image file {self.img_path}')
+		self.do_create(self.size, self.img_path)
+		atexit.register(self.try_detach)
+
+	def attach(self, dev_mode=None, silent=False):
+		if res := self._get_associations():
+			die(2, f'Device{suf(res)} {fmt_list(res,fmt="barest")} already associated with {self.img_path}')
+		dev = self.get_new_dev()
+		if dev_mode:
+			self.dev_mode_orig = '0{:o}'.format(os.stat(dev).st_mode & 0xfff)
+			if not silent:
+				imsg(f'Changing permissions on device {dev} to {dev_mode!r}')
+			do_run(['sudo', 'chmod', dev_mode, dev])
+		if not silent:
+			imsg(f'Attaching {dev or self.img_path!r}')
+		self.do_attach(self.img_path, dev)
+
+	def detach(self, silent=False):
+		dev = self.dev
+		if not silent:
+			imsg(f'Detaching device {dev!r}')
+		self.do_detach(dev)
+		if hasattr(self, 'dev_mode_orig'):
+			if not silent:
+				imsg(f'Resetting permissions on device {dev} to {self.dev_mode_orig!r}')
+			do_run(['sudo', 'chmod', self.dev_mode_orig, dev])
+			delattr(self, 'dev_mode_orig')
+
+	def __del__(self):
+		self.try_detach()
+
+class VirtBlockDeviceLinux(VirtBlockDeviceBase):
+
+	def __init__(self, img_path, size):
+		self.img_path = Path(img_path).resolve()
+		self.size = size
+
+	def _get_associations(self):
+		cmd = ['/sbin/losetup', '-n', '-O', 'NAME', '-j', str(self.img_path)]
+		return do_run(cmd).stdout.decode().splitlines()
+
+	def get_new_dev(self):
+		return do_run(['sudo', '/sbin/losetup', '-f']).stdout.decode().strip()
+
+	def do_create(self, size, path):
+		do_run(['truncate', f'--size={size}', str(path)])
+
+	def do_attach(self, path, dev):
+		do_run(['sudo', '/sbin/losetup', dev, str(path)])
+
+	def do_detach(self, dev, check=True):
+		do_run(['/sbin/losetup', '-d', dev], check=check)