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, read_file

import gi
gi.require_version('GLib', '2.0')
gi.require_version('BlockDev', '3.0')
from gi.repository import GLib, BlockDev


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"

    def setUp(self):
        self.addCleanup(self._clean_up)

    def _clean_up(self):
        self.log = ""
        BlockDev.utils_set_log_level(BlockDev.UTILS_LOG_WARNING)

    @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)

        # set default log level to info which is used for exec calls
        BlockDev.utils_set_log_level(BlockDev.UTILS_LOG_INFO)

        succ = BlockDev.utils_exec_and_report_error(["true"])
        self.assertTrue(succ)

        succ = BlockDev.utils_exec_and_capture_output_no_progress(["true"])
        self.assertTrue(succ)

        with self.assertRaisesRegex(GLib.GError, r"Process reported exit code 1"):
            succ = BlockDev.utils_exec_and_report_error(["/bin/false"])

        succ, out, stderr, status = BlockDev.utils_exec_and_capture_output_no_progress(["/bin/false"])
        self.assertTrue(succ)
        self.assertEqual(status, 1)

        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_logging_level(self):
        succ = BlockDev.utils_init_logging(self.my_log_func)
        self.assertTrue(succ)

        # default log level should be warning, info should be ignored
        BlockDev.utils_log(BlockDev.UTILS_LOG_INFO, "info message")
        self.assertFalse(self.log)

        BlockDev.utils_log(BlockDev.UTILS_LOG_WARNING, "warning message")
        self.assertIn("warning message", self.log)

        # switch default to info
        BlockDev.utils_set_log_level(BlockDev.UTILS_LOG_INFO)
        BlockDev.utils_log(BlockDev.UTILS_LOG_INFO, "info message")
        self.assertIn("info message", 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/fake_utils/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(.*)"))

    @tag_test(TestTags.NOSTORAGE, TestTags.CORE)
    def test_exec_locale(self):
        """Verify that setting locale for exec functions works as expected"""

        succ, out = BlockDev.utils_exec_and_capture_output(["locale"])
        self.assertTrue(succ)
        self.assertIn("LC_ALL=C", out)

    @tag_test(TestTags.NOSTORAGE, TestTags.CORE)
    def test_exec_buffer_bloat(self):
        """Verify that very large output from a command is handled properly"""

        # easy 64kB of data
        cnt = 65536
        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; done" % cnt])
        self.assertTrue(succ)
        self.assertEqual(len(out), cnt)

        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", r"for i in {1..%d}; do echo -n .; echo -n \# >&2; done" % cnt])
        self.assertTrue(succ)
        self.assertEqual(len(out), cnt)

        # now exceed the system pipe buffer size
        # pipe(7): The maximum size (in bytes) of individual pipes that can be set by users without the CAP_SYS_RESOURCE capability.
        cnt = int(read_file("/proc/sys/fs/pipe-max-size")) + 100
        self.assertGreater(cnt, 0)

        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; done" % cnt])
        self.assertTrue(succ)
        self.assertEqual(len(out), cnt)

        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", r"for i in {1..%d}; do echo -n .; echo -n \# >&2; done" % cnt])
        self.assertTrue(succ)
        self.assertEqual(len(out), cnt)

        # make use of some newlines
        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; if [ $(($i%%500)) -eq 0 ]; then echo ''; fi; done" % cnt])
        self.assertTrue(succ)
        self.assertGreater(len(out), cnt)

        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", r"for i in {1..%d}; do echo -n .; echo -n \# >&2; if [ $(($i%%500)) -eq 0 ]; then echo ''; echo '' >&2; fi; done" % cnt])
        self.assertTrue(succ)
        self.assertGreater(len(out), cnt)

        (succ, out, stderr, status)  = BlockDev.utils_exec_and_capture_output_no_progress(["bash", "-c", r"for i in {1..%d}; do echo -n .; echo -n \# >&2; if [ $(($i%%500)) -eq 0 ]; then echo ''; echo '' >&2; fi; done" % cnt])
        self.assertTrue(succ)
        self.assertGreater(len(out), cnt)

        (succ, out, stderr, status) = BlockDev.utils_exec_and_capture_output_no_progress(["bash", "-c", r"for i in {1..%d}; do echo -n . >&2; echo -n \# >&2; if [ $(($i%%500)) -eq 0 ]; then echo '' >&2; fi; done; exit 123" % cnt])
        self.assertTrue(succ)
        self.assertEqual(status, 123)
        self.assertEqual(len(out), 0)
        self.assertGreater(len(stderr), cnt)

    EXEC_PROGRESS_MSG = "Aloha, I'm the progress line you should match."

    def my_exec_progress_func_concat(self, line):
        """Expect an concatenated string"""
        s = ""
        for i in range(10):
            s += self.EXEC_PROGRESS_MSG
        self.assertEqual(line, s)
        self.num_matches += 1
        return 0

    def my_exec_progress_func(self, line):
        self.assertTrue(re.match(r".*%s.*" % self.EXEC_PROGRESS_MSG, line))
        self.num_matches += 1
        return 0

    def test_exec_buffer_bloat_progress(self):
        """Verify that progress reporting can handle large output"""

        # no newlines, should match just a single occurrence on EOF
        cnt = 10
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..10}; do echo -n \"%s\"; done" % self.EXEC_PROGRESS_MSG], None, self.my_exec_progress_func_concat)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, 1)

        # ten matches
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt)

        # 100k matches
        cnt = 100000
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt)

        # 100k matches on stderr
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt)

        # 100k matches on stderr and stdout
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt * 2)

        # stdout and stderr output, non-zero return from the command and very long exception message
        self.num_matches = 0
        with self.assertRaisesRegex(GLib.GError, r"Process reported exit code 66"):
            status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done; exit 66" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertEqual(self.num_matches, cnt * 2)

        # no progress reporting callback given, tests slightly different code path
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, None)
        self.assertTrue(status)

    def test_exec_null_bytes(self):
        """Verify that null bytes in the output are processed properly"""

        TEST_DATA = ["First line", "Second line", "Third line"]

        status, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "echo -e \"%s\\0%s\\n%s\"" % (TEST_DATA[0], TEST_DATA[1], TEST_DATA[2])])
        self.assertTrue(status)
        self.assertIn(TEST_DATA[0], out)
        self.assertIn(TEST_DATA[1], out)
        self.assertIn(TEST_DATA[2], out)
        self.assertNotIn("kuku!", out)

        # ten matches
        cnt = 10
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt * 2)

        # 100k matches
        cnt = 100000
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt * 2)

        # 100k matches on stderr
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt * 2)

        # 100k matches on stderr and stdout
        self.num_matches = 0
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\"; echo -e \"%s\\0%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertTrue(status)
        self.assertEqual(self.num_matches, cnt * 4)

        # stdout and stderr output, non-zero return from the command and very long exception message
        self.num_matches = 0
        with self.assertRaisesRegex(GLib.GError, r"Process reported exit code 66"):
            status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\"; echo -e \"%s\\0%s\" >&2; done; exit 66" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)
        self.assertEqual(self.num_matches, cnt * 4)

        # no progress reporting callback given, tests slightly different code path
        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo -e \"%s\\0%s\"; echo -e \"%s\\0%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, None)
        self.assertTrue(status)

    def test_exec_large_input(self):
        """Verify that large input is passed to the process properly"""

        DATA_MATCH = "==END=="

        data = "Begin data =="
        for i in range(6666666):
            data += "bloat"
        data += DATA_MATCH

        # command is not accepting stdin, write() will return an error
        with self.assertRaisesRegex(GLib.GError, r"Failed to write to stdin of the process: Broken pipe"):
            BlockDev.utils_exec_with_input(["false"], data, None)

        # test that `grep` returns error on non-match
        with self.assertRaisesRegex(GLib.GError, r"Process reported exit code 1"):
            BlockDev.utils_exec_with_input(["grep", "unmatched"], data, None)

        # expecting that `grep` will find the string at the end
        status = BlockDev.utils_exec_with_input(["grep", DATA_MATCH], data, None)
        self.assertTrue(status)


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 --config \"devices {use_devicesfile = 0}\"" % self.loop_dev)
        self.assertEqual(ret, 0)
        self.addCleanup(run_command, "pvremove %s --config \"devices {use_devicesfile = 0}\"" % self.loop_dev)

        ret, _out, _err = run_command ("vgcreate utilsTestVG %s --config \"devices {use_devicesfile = 0}\"" % self.loop_dev)
        self.assertEqual(ret, 0)
        self.addCleanup(run_command, "vgremove -y utilsTestVG --config \"devices {use_devicesfile = 0}\"")

        ret, _out, _err = run_command ("lvcreate -n utilsTestLV -L 12M utilsTestVG --config \"devices {use_devicesfile = 0}\"")
        self.assertEqual(ret, 0)
        self.addCleanup(run_command, "lvremove -y utilsTestVG/utilsTestLV --config \"devices {use_devicesfile = 0}\"")

        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)


class UtilsKernelModuleTest(UtilsTestCase):
    @tag_test(TestTags.NOSTORAGE, TestTags.CORE)
    def test_have_kernel_module(self):
        """ Test checking for kernel modules """

        have = BlockDev.utils_have_kernel_module("definitely-not-existing-kernel-module")
        self.assertFalse(have)

        # loop should be everywhere, right?
        have = BlockDev.utils_have_kernel_module("loop")
        self.assertTrue(have)

        # lets check some filesystems support and compare with 'modprobe' results
        for fs in ("ext2", "ext3", "ext4", "xfs", "btrfs"):
            have_fs = BlockDev.utils_have_kernel_module(fs)
            ret, _out, _err = run_command("modprobe --dry-run %s" % fs)
            if ret == 0:
                self.assertTrue(have_fs)
            else:
                self.assertFalse(have_fs)