import unittest
import re
import os
import overrides_hack
from utils import fake_utils, create_sparse_tempfile, create_lio_device, delete_lio_device, run_command, TestTags, tag_test
from gi.repository import BlockDev, GLib
class UtilsTestCase(unittest.TestCase):
requested_plugins = []
@classmethod
def setUpClass(cls):
if not BlockDev.is_initialized():
BlockDev.init(cls.requested_plugins, None)
else:
BlockDev.reinit(cls.requested_plugins, True, None)
class UtilsExecProgressTest(UtilsTestCase):
log = []
def my_progress_func(self, task, status, completion, msg):
self.assertTrue(isinstance(completion, int))
self.log.append(completion)
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_initialization(self):
""" Verify that progress report can (de)initialized"""
succ = BlockDev.utils_prog_reporting_initialized()
self.assertFalse(succ)
succ = BlockDev.utils_init_prog_reporting(self.my_progress_func)
self.assertTrue(succ)
succ = BlockDev.utils_prog_reporting_initialized()
self.assertTrue(succ)
succ = BlockDev.utils_init_prog_reporting(None)
self.assertTrue(succ)
succ = BlockDev.utils_prog_reporting_initialized()
self.assertFalse(succ)
class UtilsExecLoggingTest(UtilsTestCase):
log = ""
def my_log_func(self, level, msg):
# not much to verify here
self.assertTrue(isinstance(level, int))
self.assertTrue(isinstance(msg, str))
self.log += msg + "\n"
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_logging(self):
"""Verify that setting up and using exec logging works as expected"""
succ = BlockDev.utils_init_logging(self.my_log_func)
self.assertTrue(succ)
succ = BlockDev.utils_exec_and_report_error(["true"])
self.assertTrue(succ)
succ, out = BlockDev.utils_exec_and_capture_output(["echo", "hi"])
self.assertTrue(succ)
self.assertEqual(out, "hi\n")
match = re.search(r'Running \[(\d+)\] true', self.log)
self.assertIsNot(match, None)
task_id1 = match.group(1)
match = re.search(r'Running \[(\d+)\] echo hi', self.log)
self.assertIsNot(match, None)
task_id2 = match.group(1)
self.assertIn("...done [%s] (exit code: 0)" % task_id1, self.log)
self.assertIn("stdout[%s]:" % task_id1, self.log)
self.assertIn("stderr[%s]:" % task_id1, self.log)
self.assertIn("stdout[%s]: hi" % task_id2, self.log)
self.assertIn("stderr[%s]:" % task_id2, self.log)
self.assertIn("...done [%s] (exit code: 0)" % task_id2, self.log)
# reset logging -> nothing more should appear in the log
succ = BlockDev.utils_init_logging(None)
self.assertTrue(succ)
old_log = self.log
succ = BlockDev.utils_exec_and_report_error(["true"])
self.assertTrue(succ)
self.assertEqual(old_log, self.log)
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_version_cmp(self):
"""Verify that version comparison works as expected"""
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("malformed", "1.0")
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("1.0", "malformed")
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("1,0", "1.0")
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("1.0", "1,0")
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("1.x.0", "1.0")
with self.assertRaises(GLib.GError):
BlockDev.utils_version_cmp("1.0", "1.x.0")
self.assertEqual(BlockDev.utils_version_cmp("1", "1"), 0)
self.assertEqual(BlockDev.utils_version_cmp("1.0", "1.0"), 0)
self.assertEqual(BlockDev.utils_version_cmp("1.0.1", "1.0.1"), 0)
self.assertEqual(BlockDev.utils_version_cmp("1.0.1-1", "1.0.1-1"), 0)
self.assertEqual(BlockDev.utils_version_cmp("1.1", "1"), 1)
self.assertEqual(BlockDev.utils_version_cmp("1.1", "1.0"), 1)
self.assertEqual(BlockDev.utils_version_cmp("1.1.1", "1.1"), 1)
self.assertEqual(BlockDev.utils_version_cmp("1.1.1-1", "1.1.1"), 1)
self.assertEqual(BlockDev.utils_version_cmp("1.2", "1.1.2"), 1)
self.assertEqual(BlockDev.utils_version_cmp("1", "1.1"), -1)
self.assertEqual(BlockDev.utils_version_cmp("1.0", "1.1"), -1)
self.assertEqual(BlockDev.utils_version_cmp("1.1", "1.1.1"), -1)
self.assertEqual(BlockDev.utils_version_cmp("1.1.1", "1.1.1-1"), -1)
self.assertEqual(BlockDev.utils_version_cmp("1.1.2", "1.2"), -1)
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_util_version(self):
"""Verify that checking utility availability works as expected"""
with self.assertRaises(GLib.GError):
BlockDev.utils_check_util_version("libblockdev-fake-util", None, None, None)
with fake_utils("tests/utils_fake_util/"):
with self.assertRaises(GLib.GError):
# with no argument, the output is "Version: 1.2" which is not a
# valid version without regexp
BlockDev.utils_check_util_version("libblockdev-fake-util", "1.3", "", None)
with self.assertRaises(GLib.GError):
# libblockdev-fake-util with no arguments reports 1.2 which is too low
BlockDev.utils_check_util_version("libblockdev-fake-util", "1.3", "", "Version:\\s(.*)")
# just what we require
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util", "1.2", "", "Version:\\s(.*)"))
with self.assertRaises(GLib.GError):
# libblockdev-fake-util with "version" reports 1.1 which is too low
BlockDev.utils_check_util_version("libblockdev-fake-util", "1.2", "version", "Version:\\s(.*)")
# just what we require
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util", "1.1", "version", "Version:\\s(.*)"))
with self.assertRaises(GLib.GError):
# libblockdev-fake-util with "--version" reports 1.0 which is too low
BlockDev.utils_check_util_version("libblockdev-fake-util", "1.1", "--version", None)
# just what we require
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util", "1.0", "--version", None))
# lower required version
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util", "0.9", "--version", None))
# version on stderr
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util-stderr", "1.1", "version", "Version:\\s(.*)"))
# exit code != 0
self.assertTrue(BlockDev.utils_check_util_version("libblockdev-fake-util-fail", "1.1", "version", "Version:\\s(.*)"))
class UtilsDevUtilsTestCase(UtilsTestCase):
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_resolve_device(self):
"""Verify that resolving device spec works as expected"""
with self.assertRaises(GLib.GError):
BlockDev.utils_resolve_device("no_such_device")
dev = "/dev/libblockdev-test-dev"
with open(dev, "w"):
pass
self.addCleanup(os.unlink, dev)
# full path, no symlink, should just return the same
self.assertEqual(BlockDev.utils_resolve_device(dev), dev)
# just the name of the device, should return the full path
self.assertEqual(BlockDev.utils_resolve_device(dev[5:]), dev)
dev_dir = "/dev/libblockdev-test-dir"
os.mkdir(dev_dir)
self.addCleanup(os.rmdir, dev_dir)
dev_link = dev_dir + "/test-dev-link"
os.symlink("../" + dev[5:], dev_link)
self.addCleanup(os.unlink, dev_link)
# should resolve the symlink
self.assertEqual(BlockDev.utils_resolve_device(dev_link), dev)
# should resolve the symlink even without the "/dev" prefix
self.assertEqual(BlockDev.utils_resolve_device(dev_link[5:]), dev)
class UtilsDevUtilsSymlinksTestCase(UtilsTestCase):
def setUp(self):
self.addCleanup(self._clean_up)
self.dev_file = create_sparse_tempfile("lvm_test", 1024**3)
try:
self.loop_dev = create_lio_device(self.dev_file)
except RuntimeError as e:
raise RuntimeError("Failed to setup loop device for testing: %s" % e)
def _clean_up(self):
try:
delete_lio_device(self.loop_dev)
except RuntimeError:
# just move on, we can do no better here
pass
os.unlink(self.dev_file)
@tag_test(TestTags.CORE)
def test_get_device_symlinks(self):
"""Verify that getting device symlinks works as expected"""
with self.assertRaises(GLib.GError):
BlockDev.utils_get_device_symlinks("no_such_device")
symlinks = BlockDev.utils_get_device_symlinks(self.loop_dev)
# there should be at least 2 symlinks for something like "/dev/sda" (in /dev/disk/by-id/)
self.assertGreaterEqual(len(symlinks), 2)
symlinks = BlockDev.utils_get_device_symlinks(self.loop_dev[5:])
self.assertGreaterEqual(len(symlinks), 2)
# create an LV to get a device with more symlinks
ret, _out, _err = run_command ("pvcreate %s" % self.loop_dev)
self.assertEqual(ret, 0)
self.addCleanup(run_command, "pvremove %s" % self.loop_dev)
ret, _out, _err = run_command ("vgcreate utilsTestVG %s" % self.loop_dev)
self.assertEqual(ret, 0)
self.addCleanup(run_command, "vgremove -y utilsTestVG")
ret, _out, _err = run_command ("lvcreate -n utilsTestLV -L 12M utilsTestVG")
self.assertEqual(ret, 0)
self.addCleanup(run_command, "lvremove -y utilsTestVG/utilsTestLV")
symlinks = BlockDev.utils_get_device_symlinks("utilsTestVG/utilsTestLV")
# there should be at least 4 symlinks for an LV
self.assertGreaterEqual(len(symlinks), 4)
class UtilsLinuxKernelVersionTest(UtilsTestCase):
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_initialization(self):
""" Test Linux kernel version detection"""
ver = BlockDev.utils_get_linux_version()
self.assertGreater(ver.major, 0)
ret = BlockDev.utils_check_linux_version(ver.major, ver.minor, ver.micro)
self.assertEqual(ret, 0)
ret = BlockDev.utils_check_linux_version(ver.major - 1, ver.minor, ver.micro)
self.assertGreater(ret, 0)
ret = BlockDev.utils_check_linux_version(ver.major + 1, ver.minor, ver.micro)
self.assertLess(ret, 0)
ver2 = BlockDev.utils_get_linux_version()
self.assertEqual(ver.major, ver2.major)
self.assertEqual(ver.minor, ver2.minor)
self.assertEqual(ver.micro, ver2.micro)