diff --git a/tests/smart_test.py b/tests/smart_test.py index 39c91779..13a0f931 100644 --- a/tests/smart_test.py +++ b/tests/smart_test.py @@ -1,10 +1,8 @@ import unittest import os -import glob -import time import overrides_hack -from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, TestTags, tag_test, write_file, run_command, required_plugins +from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, TestTags, tag_test, required_plugins, setup_scsi_debug, clean_scsi_debug import gi gi.require_version('GLib', '2.0') @@ -59,29 +57,6 @@ def _clean_loop(self): pass os.unlink(self.loop_dev_file) - def _setup_scsi_debug(self): - res, _out, _err = run_command('modprobe scsi_debug') - self.assertEqual(res, 0) - dirs = [] - while len(dirs) < 1: - dirs = glob.glob('/sys/bus/pseudo/drivers/scsi_debug/adapter*/host*/target*/*:*/block') - time.sleep(0.1) - self.scsi_debug_dev = os.listdir(dirs[0]) - self.assertEqual(len(self.scsi_debug_dev), 1) - self.scsi_debug_dev = '/dev/' + self.scsi_debug_dev[0] - self.assertTrue(os.path.exists(self.scsi_debug_dev)) - - def _clean_scsi_debug(self): - try: - device = self.scsi_debug_dev.split('/')[-1] - if os.path.exists('/sys/block/' + device): - self.write_file('/sys/block/%s/device/delete' % device, '1') - while os.path.exists(device): - time.sleep(0.1) - self.run_command('modprobe -r scsi_debug') - except: - pass - @tag_test(TestTags.NOSTORAGE) def test_plugin_version(self): self.assertEqual(BlockDev.get_plugin_soname(BlockDev.Plugin.SMART), "libbd_smart.so.3") @@ -110,8 +85,8 @@ def test_ata_info(self): BlockDev.smart_ata_get_info(self.loop_dev) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_ata_get_info(self.scsi_debug_dev) @@ -144,8 +119,8 @@ def test_smart_enable_disable(self): BlockDev.smart_set_enabled(self.loop_dev, True) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_set_enabled(self.scsi_debug_dev, False) with self.assertRaisesRegex(GLib.GError, msg): @@ -182,8 +157,8 @@ def test_smart_selftest(self): BlockDev.smart_device_self_test(self.loop_dev, t) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) for t in tests: with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_device_self_test(self.scsi_debug_dev, t) @@ -199,6 +174,7 @@ def test_scsi_info(self): # LIO device (SCSI) self._setup_lio() + self.addCleanup(self._clean_lio) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_scsi_get_info(self.lio_dev) @@ -209,8 +185,8 @@ def test_scsi_info(self): BlockDev.smart_scsi_get_info(self.loop_dev) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_scsi_get_info(self.scsi_debug_dev) diff --git a/tests/smartmontools_test.py b/tests/smartmontools_test.py index eee0cffe..91c370c9 100644 --- a/tests/smartmontools_test.py +++ b/tests/smartmontools_test.py @@ -1,11 +1,9 @@ import unittest import os -import glob -import time import shutil import overrides_hack -from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, fake_utils, TestTags, tag_test, write_file, run_command, required_plugins +from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, fake_utils, TestTags, tag_test, required_plugins, setup_scsi_debug, clean_scsi_debug import gi gi.require_version('GLib', '2.0') @@ -59,29 +57,6 @@ def _clean_loop(self): pass os.unlink(self.loop_dev_file) - def _setup_scsi_debug(self): - res, _out, _err = run_command('modprobe scsi_debug') - self.assertEqual(res, 0) - dirs = [] - while len(dirs) < 1: - dirs = glob.glob('/sys/bus/pseudo/drivers/scsi_debug/adapter*/host*/target*/*:*/block') - time.sleep(0.1) - self.scsi_debug_dev = os.listdir(dirs[0]) - self.assertEqual(len(self.scsi_debug_dev), 1) - self.scsi_debug_dev = '/dev/' + self.scsi_debug_dev[0] - self.assertTrue(os.path.exists(self.scsi_debug_dev)) - - def _clean_scsi_debug(self): - try: - device = self.scsi_debug_dev.split('/')[-1] - if os.path.exists('/sys/block/' + device): - self.write_file('/sys/block/%s/device/delete' % device, '1') - while os.path.exists(device): - time.sleep(0.1) - self.run_command('modprobe -r scsi_debug') - except: - pass - @tag_test(TestTags.NOSTORAGE) def test_plugin_version(self): self.assertEqual(BlockDev.get_plugin_soname(BlockDev.Plugin.SMART), "libbd_smartmontools.so.3") @@ -132,8 +107,8 @@ def test_ata_info(self): BlockDev.smart_ata_get_info(self.loop_dev, [BlockDev.ExtraArg.new("--device=ata", "")]) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) msg = r"Error parsing smartctl JSON data: The member .ata_smart_data. is not defined in the object at the current position." with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_ata_get_info(self.scsi_debug_dev, None) @@ -280,8 +255,8 @@ def test_smart_enable_disable(self): BlockDev.smart_set_enabled(self.loop_dev, True, [BlockDev.ExtraArg.new("--device=ata", "")]) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) BlockDev.smart_set_enabled(self.scsi_debug_dev, False) BlockDev.smart_set_enabled(self.scsi_debug_dev, True) BlockDev.smart_set_enabled(self.scsi_debug_dev, False, [BlockDev.ExtraArg.new("--device=scsi", "")]) @@ -344,8 +319,8 @@ def test_smart_selftest(self): BlockDev.smart_device_self_test(self.loop_dev, t, [BlockDev.ExtraArg.new("--device=scsi", "")]) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) for t in [BlockDev.SmartSelfTestOp.OFFLINE, BlockDev.SmartSelfTestOp.SHORT, BlockDev.SmartSelfTestOp.LONG, BlockDev.SmartSelfTestOp.CONVEYANCE, BlockDev.SmartSelfTestOp.ABORT]: @@ -374,6 +349,7 @@ def test_scsi_info(self): # LIO device (SCSI) self._setup_lio() + self.addCleanup(self._clean_lio) msg = r"Error getting SCSI SMART info: Some SMART or other ATA command to the disk failed, or there was a checksum error in a SMART data structure." with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_scsi_get_info(self.lio_dev, None) @@ -396,8 +372,8 @@ def test_scsi_info(self): BlockDev.smart_scsi_get_info(self.loop_dev, [BlockDev.ExtraArg.new("--device=ata", "")]) # scsi_debug - self._setup_scsi_debug() - self.addCleanup(self._clean_scsi_debug) + self.scsi_debug_dev = setup_scsi_debug() + self.addCleanup(clean_scsi_debug, self.scsi_debug_dev) msg = r"Error getting SCSI SMART info: Some SMART or other ATA command to the disk failed, or there was a checksum error in a SMART data structure." with self.assertRaisesRegex(GLib.GError, msg): BlockDev.smart_scsi_get_info(self.scsi_debug_dev, None) diff --git a/tests/utils.py b/tests/utils.py index 0889862a..cf0d9125 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -231,6 +231,49 @@ def delete_lio_device(dev_path): else: raise RuntimeError("Unknown device '%s'" % dev_path) + +def setup_scsi_debug(): + res, out, err = run_command('modprobe scsi_debug') + if res != 0: + raise RuntimeError("Failed to load scsi_debug module: %s %s" % (out, err)) + dirs = [] + n_tries = 0 + while len(dirs) < 1 and n_tries < 5: + dirs = glob.glob('/sys/bus/pseudo/drivers/scsi_debug/adapter*/host*/target*/*:*/block') + time.sleep(0.5) + n_tries += 1 + if len(dirs) < 1: + raise RuntimeError("Failed to setup SCSI debug device for testing") + scsi_debug_dev = os.listdir(dirs[0]) + if len(scsi_debug_dev) != 1: + raise RuntimeError("Failed to setup SCSI debug device for testing") + + scsi_debug_dev = '/dev/' + scsi_debug_dev[0] + if not os.path.exists(scsi_debug_dev): + raise RuntimeError("Failed to setup SCSI debug device for testing") + + return scsi_debug_dev + + +def clean_scsi_debug(scsi_debug_dev): + try: + device = scsi_debug_dev.split('/')[-1] + if os.path.exists('/sys/block/' + device): + write_file('/sys/block/%s/device/delete' % device, '1') + n_tries = 0 + while os.path.exists(device) and n_tries < 5: + time.sleep(0.5) + n_tries += 1 + n_tries = 0 + while n_tries < 5: + res, _out, _err = run_command('modprobe -r scsi_debug') + if res == 0: + break + time.sleep(0.5) + n_tries += 1 + except: + pass + def find_nvme_ctrl_devs_for_subnqn(subnqn): """ Find NVMe controller devices for the specified subsystem nqn