import unittest
import os
import re
import time
from contextlib import contextmanager
import overrides_hack
import six

from utils import create_sparse_tempfile, create_lio_device, delete_lio_device, fake_utils, fake_path, skip_on
from gi.repository import BlockDev, GLib


@contextmanager
def wait_for_action(action_name):
    try:
        yield
    finally:
        time.sleep(2)
        action = True
        while action:
            with open("/proc/mdstat", "r") as f:
                action = action_name in f.read()
            if action:
                print("Sleeping")
                time.sleep(1)

class MDTest(unittest.TestCase):

    requested_plugins = BlockDev.plugin_specs_from_names(("mdraid",))

    @classmethod
    def setUpClass(cls):
        if not BlockDev.is_initialized():
            BlockDev.init(cls.requested_plugins, None)
        else:
            BlockDev.reinit(cls.requested_plugins, True, None)

class MDNoDevTestCase(MDTest):

    requested_plugins = BlockDev.plugin_specs_from_names(("mdraid",))

    @classmethod
    def setUpClass(cls):
        if not BlockDev.is_initialized():
            BlockDev.init(cls.requested_plugins, None)
        else:
            BlockDev.reinit(cls.requested_plugins, True, None)

    def test_get_superblock_size(self):
        """Verify that superblock size si calculated properly"""

        # 2 MiB for versions <= 1.0
        self.assertEqual(BlockDev.md_get_superblock_size(2 * 1024**3, "0.9"), 2 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(2 * 1024**3, "1.0"), 2 * 1024**2)

        # no version, "default" or > 1.0
        self.assertEqual(BlockDev.md_get_superblock_size(256 * 1024**3, None), 128 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(128 * 1024**3, None), 128 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(64 * 1024**3, "default"), 64 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(63 * 1024**3, "default"), 32 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(10 * 1024**3, "1.1"), 8 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(1 * 1024**3, "1.1"), 1 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(1023 * 1024**2, "1.2"), 1 * 1024**2)
        self.assertEqual(BlockDev.md_get_superblock_size(512 * 1024**2, "1.2"), 1 * 1024**2)

        # unsupported version -> default superblock size
        self.assertEqual(BlockDev.md_get_superblock_size(257 * 1024**2, version="unknown version"),
                         2 * 1024**2)

    def test_canonicalize_uuid(self):
        """Verify that UUID canonicalization works as expected"""

        self.assertEqual(BlockDev.md_canonicalize_uuid("3386ff85:f5012621:4a435f06:1eb47236"),
                         "3386ff85-f501-2621-4a43-5f061eb47236")

        with six.assertRaisesRegex(self, GLib.GError, r'malformed or invalid'):
            BlockDev.md_canonicalize_uuid("malformed-uuid-example")

    def test_get_md_uuid(self):
        """Verify that getting UUID in MD RAID format works as expected"""

        self.assertEqual(BlockDev.md_get_md_uuid("3386ff85-f501-2621-4a43-5f061eb47236"),
                         "3386ff85:f5012621:4a435f06:1eb47236")

        with six.assertRaisesRegex(self, GLib.GError, r'malformed or invalid'):
            BlockDev.md_get_md_uuid("malformed-uuid-example")

class MDTestCase(MDTest):

    def setUp(self):
        if os.uname()[-1] == "i686":
            self.skipTest("Skipping hanging MD RAID tests on i686")

        self.addCleanup(self._clean_up)
        self.dev_file = create_sparse_tempfile("md_test", 10 * 1024**2)
        self.dev_file2 = create_sparse_tempfile("md_test", 10 * 1024**2)
        self.dev_file3 = create_sparse_tempfile("md_test", 10 * 1024**2)

        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)
        try:
            self.loop_dev2 = create_lio_device(self.dev_file2)
        except RuntimeError as e:
            raise RuntimeError("Failed to setup loop device for testing: %s" % e)
        try:
            self.loop_dev3 = create_lio_device(self.dev_file3)
        except RuntimeError as e:
            raise RuntimeError("Failed to setup loop device for testing: %s" % e)

    def _clean_up(self):
        try:
            BlockDev.md_deactivate("bd_test_md")
        except:
            pass
        try:
            BlockDev.md_deactivate(BlockDev.md_node_from_name("bd_test_md"))
        except:
            pass
        try:
            BlockDev.md_destroy(self.loop_dev)
        except:
            pass
        try:
            BlockDev.md_destroy(self.loop_dev2)
        except:
            pass
        try:
            BlockDev.md_destroy(self.loop_dev3)
        except:
            pass
        try:
            BlockDev.md_deactivate("bd_test_md")
        except:
            pass
        try:
            BlockDev.md_deactivate(BlockDev.md_node_from_name("bd_test_md"))
        except:
            pass

        try:
            delete_lio_device(self.loop_dev)
        except RuntimeError:
            # just move on, we can do no better here
            pass
        os.unlink(self.dev_file)

        try:
            delete_lio_device(self.loop_dev2)
        except RuntimeError:
            # just move on, we can do no better here
            pass
        os.unlink(self.dev_file2)

        try:
            delete_lio_device(self.loop_dev3)
        except RuntimeError:
            # just move on, we can do no better here
            pass
        os.unlink(self.dev_file3)


class MDTestCreateDeactivateDestroy(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_create_deactivate_destroy(self):
        """Verify that it is possible to create, deactivate and destroy an MD RAID"""

        with self.assertRaises(GLib.GError):
            BlockDev.md_create("bd_test_md2", "raid1",
                               ["/non/existing/device", self.loop_dev2],
                               1, None, True)

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        # newly created array should be 'clean'
        state = BlockDev.md_get_status("bd_test_md")
        self.assertEqual(state, "clean")

        succ = BlockDev.md_deactivate("bd_test_md")
        self.assertTrue(succ)

        succ = BlockDev.md_destroy(self.loop_dev)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev2)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev3)
        self.assertTrue(succ)

class MDTestCreateWithChunkSize(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_create_with_chunk_size(self):
        """Verify that it is possible to create and MD RAID with specific chunk size """

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid0",
                                      [self.loop_dev, self.loop_dev2],
                                      0, None, False, 512 * 1024)
            self.assertTrue(succ)

        ex_data = BlockDev.md_examine(self.loop_dev)
        self.assertEqual(ex_data.chunk_size, 512 * 1024)

        succ = BlockDev.md_deactivate("bd_test_md")
        self.assertTrue(succ)

        succ = BlockDev.md_destroy(self.loop_dev)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev2)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev3)
        self.assertTrue(succ)

class MDTestActivateDeactivate(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_activate_deactivate(self):
        """Verify that it is possible to activate and deactivate an MD RAID"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        with self.assertRaises(GLib.GError):
            BlockDev.md_deactivate("non_existing_md")

        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate("bd_test_md")
            self.assertTrue(succ)

        with self.assertRaises(GLib.GError):
            BlockDev.md_activate("bd_test_md",
                                 ["/non/existing/device", self.loop_dev2, self.loop_dev3], None)

        with wait_for_action("resync"):
            succ = BlockDev.md_activate("bd_test_md",
                                        [self.loop_dev, self.loop_dev2, self.loop_dev3], None)
            self.assertTrue(succ)

        # try to deactivate using the node instead of name
        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate(BlockDev.md_node_from_name("bd_test_md"))
            self.assertTrue(succ)

        # try to activate using full path, not just the name
        # (it should work too and blivet does this)
        with wait_for_action("resync"):
            succ = BlockDev.md_activate("/dev/md/bd_test_md",
                                        [self.loop_dev, self.loop_dev2, self.loop_dev3], None)
            self.assertTrue(succ)

class MDTestActivateWithUUID(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_activate_with_uuid(self):
        """Verify that it is possible to activate an MD RAID with UUID"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate("bd_test_md")
            self.assertTrue(succ)

        md_info = BlockDev.md_examine(self.loop_dev)
        self.assertTrue(md_info)
        self.assertTrue(md_info.uuid)

        with wait_for_action("resync"):
            succ = BlockDev.md_activate("bd_test_md", [self.loop_dev, self.loop_dev2, self.loop_dev3], md_info.uuid)

class MDTestActivateByUUID(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_activate_by_uuid(self):
        """Verify that it is possible to activate an MD RAID by UUID"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate("bd_test_md")
            self.assertTrue(succ)

        md_info = BlockDev.md_examine(self.loop_dev)
        self.assertTrue(md_info)
        self.assertTrue(md_info.uuid)

        # should work with member devices specified
        with wait_for_action("resync"):
            succ = BlockDev.md_activate(None, [self.loop_dev, self.loop_dev2, self.loop_dev3], md_info.uuid)

        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate("bd_test_md")
            self.assertTrue(succ)

        # as well as without them
        with wait_for_action("resync"):
            succ = BlockDev.md_activate(None, None, md_info.uuid)


class MDTestNominateDenominate(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_nominate_denominate(self):
        """Verify that it is possible to nominate and denominate an MD RAID device"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, False)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_denominate(self.loop_dev)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_nominate(self.loop_dev)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_denominate(self.loop_dev)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_nominate(self.loop_dev)
            self.assertTrue(succ)

        with wait_for_action("resync"):
            succ = BlockDev.md_deactivate(BlockDev.md_node_from_name("bd_test_md"))
            self.assertTrue(succ)

class MDTestNominateDenominateActive(MDTestCase):
    # slow and leaking an MD array because with a nominated spare device, it
    # cannot be deactivated in the end (don't ask me why)
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    @unittest.skipIf("JENKINS_HOME" in os.environ, "skipping leaky test in jenkins")
    @unittest.skipUnless("FEELINGLUCKY" in os.environ, "skipping, not feeling lucky")
    def test_nominate_denominate_active(self):
        """Verify that nominate and denominate deivice works as expected on (de)activated MD RAID"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, False)
            self.assertTrue(succ)

        # can not re-add in incremental mode because the array is active
        with self.assertRaises(GLib.GError):
            BlockDev.md_nominate(self.loop_dev3)

        succ = BlockDev.md_deactivate("bd_test_md");
        self.assertTrue(succ)

        # once the array is deactivated, can add in incremental mode
        succ = BlockDev.md_nominate(self.loop_dev3)
        self.assertTrue(succ)

        # cannot re-add twice
        with self.assertRaises(GLib.GError):
            succ = BlockDev.md_nominate(self.loop_dev3)
            self.assertTrue(succ)

class MDTestAddRemove(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    @skip_on("debian", reason="Removing spare disks from an array is broken on Debian")
    def test_add_remove(self):
        """Verify that it is possible to add a device to and remove from an MD RAID"""

        # the MD array doesn't exist yet
        with self.assertRaises(GLib.GError):
            BlockDev.md_add("bd_test_md", self.loop_dev3, 0, None)

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2],
                                      0, None, False)
            self.assertTrue(succ)

        with self.assertRaises(GLib.GError):
            BlockDev.md_add("bd_test_md", "/non/existing/device", 0, None)

        # add the device as a spare
        succ = BlockDev.md_add("bd_test_md", self.loop_dev3, 0, None)
        self.assertTrue(succ)

        md_info = BlockDev.md_detail("bd_test_md")
        self.assertEqual(md_info.raid_devices, 2)
        self.assertEqual(md_info.spare_devices, 1)

        with self.assertRaises(GLib.GError):
            BlockDev.md_add("bd_test_md", self.loop_dev3, 0, None)

        # now remove the spare device (should be possible without --fail)
        with wait_for_action("resync"):
            succ = BlockDev.md_remove("bd_test_md", self.loop_dev3, False, None)
            self.assertTrue(succ)

        md_info = BlockDev.md_detail("bd_test_md")
        self.assertEqual(md_info.raid_devices, 2)
        self.assertEqual(md_info.spare_devices, 0)

        # remove one of the original devices (with --fail enabled)
        with wait_for_action("resync"):
            succ = BlockDev.md_remove("bd_test_md", self.loop_dev2, True, None)
            self.assertTrue(succ)

        md_info = BlockDev.md_detail("bd_test_md")
        self.assertEqual(md_info.raid_devices, 2)
        self.assertEqual(md_info.active_devices, 1)
        self.assertEqual(md_info.spare_devices, 0)

        # now try to add it back -- it should be re-added automatically as
        # a RAID device, not a spare device
        with wait_for_action("recovery"):
            succ = BlockDev.md_add("bd_test_md", self.loop_dev2, 0, None)
            self.assertTrue(succ)

        md_info = BlockDev.md_detail("bd_test_md")
        self.assertEqual(md_info.raid_devices, 2)
        self.assertEqual(md_info.active_devices, 2)
        self.assertEqual(md_info.spare_devices, 0)

class MDTestExamineDetail(MDTestCase):
    # sleeps to let MD RAID sync things
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_examine_detail(self):
        """Verify that it is possible to get info about an MD RAID"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        ex_data = BlockDev.md_examine(self.loop_dev)
        # test that we got something
        self.assertTrue(ex_data)

        # verify some known data
        self.assertEqual(ex_data.device, "/dev/md/bd_test_md")
        self.assertEqual(ex_data.level, "raid1")
        self.assertEqual(ex_data.num_devices, 2)
        self.assertTrue(ex_data.name.endswith("bd_test_md"))
        self.assertEqual(len(ex_data.metadata), 3)
        self.assertTrue(ex_data.size < (10 * 1024**2))
        self.assertTrue(re.match(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', ex_data.uuid))

        de_data = BlockDev.md_detail("bd_test_md")
        # test that we got something
        self.assertTrue(de_data)

        # verify some known data
        self.assertEqual(de_data.device, "/dev/md/bd_test_md")
        self.assertTrue(de_data.name.endswith("bd_test_md"))
        self.assertEqual(len(de_data.metadata), 3)
        self.assertEqual(de_data.level, "raid1")
        self.assertEqual(de_data.raid_devices, 2)
        self.assertEqual(de_data.total_devices, 3)
        self.assertEqual(de_data.spare_devices, 1)
        self.assertTrue(de_data.array_size < (10 * 1024**2))
        self.assertTrue(de_data.use_dev_size < (10 * 1024**2))
        if "JENKINS_HOME" not in os.environ:
            # XXX: for some reason the RAID is in "active sync" when tests run in
            # Jenkins
            self.assertTrue(de_data.clean)
        self.assertTrue(re.match(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', de_data.uuid))

        self.assertEqual(ex_data.uuid, de_data.uuid)

        # try to get detail data with some different raid specification
        node = BlockDev.md_node_from_name("bd_test_md")

        de_data = BlockDev.md_detail("/dev/md/bd_test_md")
        self.assertTrue(de_data)
        de_data = BlockDev.md_detail(node)
        self.assertTrue(de_data)
        de_data = BlockDev.md_detail("/dev/%s" % node)
        self.assertTrue(de_data)

class MDTestNameNodeBijection(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_name_node_bijection(self):
        """Verify that MD RAID node and name match each other"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        node = BlockDev.md_node_from_name("bd_test_md")
        self.assertEqual(BlockDev.md_name_from_node(node), "bd_test_md")
        self.assertEqual(BlockDev.md_name_from_node("/dev/" + node), "bd_test_md")

        with self.assertRaises(GLib.GError):
            node = BlockDev.md_node_from_name("made_up_md")

        with six.assertRaisesRegex(self, GLib.GError, r'No name'):
            BlockDev.md_name_from_node("no_such_node")

        succ = BlockDev.md_deactivate("bd_test_md");
        self.assertTrue(succ)

        succ = BlockDev.md_destroy(self.loop_dev)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev2)
        self.assertTrue(succ)
        succ = BlockDev.md_destroy(self.loop_dev3)
        self.assertTrue(succ)

class MDTestSetBitmapLocation(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_set_bitmap_location(self):
        """Verify we can change bitmap location for an existing MD array"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        succ = BlockDev.md_set_bitmap_location("bd_test_md", "none")
        self.assertTrue(succ)

        loc = BlockDev.md_get_bitmap_location("bd_test_md")
        self.assertEqual(loc, "none")

        succ = BlockDev.md_set_bitmap_location("bd_test_md", "internal")
        self.assertTrue(succ)

        loc = BlockDev.md_get_bitmap_location("bd_test_md")
        self.assertEqual(loc, "+8")

        # test some different name specifications
        # (need to switch between internal and none because setting the same
        # location multiple times results in an error)
        succ = BlockDev.md_set_bitmap_location("/dev/md/bd_test_md", "none")
        self.assertTrue(succ)

        node = BlockDev.md_node_from_name("bd_test_md")
        self.assertIsNotNone(node)

        succ = BlockDev.md_set_bitmap_location(node, "internal")
        self.assertTrue(succ)

        succ = BlockDev.md_set_bitmap_location("/dev/%s" % node, "none")
        self.assertTrue(succ)

        # get_bitmap_location should accept name, node or path
        loc = BlockDev.md_get_bitmap_location(node)
        self.assertEqual(loc, "none")

        loc = BlockDev.md_get_bitmap_location("/dev/%s" % node)
        self.assertEqual(loc, "none")

        loc = BlockDev.md_get_bitmap_location("/dev/md/bd_test_md")
        self.assertEqual(loc, "none")


class MDTestRequestSyncAction(MDTestCase):
    @unittest.skipIf("SKIP_SLOW" in os.environ, "skipping slow tests")
    def test_request_sync_action(self):
        """Verify we can request sync action on an existing MD array"""

        with wait_for_action("resync"):
            succ = BlockDev.md_create("bd_test_md", "raid1",
                                      [self.loop_dev, self.loop_dev2, self.loop_dev3],
                                      1, None, True)
            self.assertTrue(succ)

        with wait_for_action("check"):
            succ = BlockDev.md_request_sync_action("bd_test_md", "check")

        node = BlockDev.md_node_from_name("bd_test_md")
        with open("/sys/block/%s/md/last_sync_action" % node) as f:
            action = f.read().strip()
        self.assertEqual(action, "check")

class FakeMDADMutilTest(MDTest):
    # no setUp nor tearDown needed, we are gonna use fake utils
    def test_fw_raid_uppercase_examine(self):
        """Verify that md_examine works with output using "RAID" instead of "Raid" and other quirks """

        with fake_utils("tests/mdadm_fw_RAID_examine"):
            ex_data = BlockDev.md_examine("fake_dev")

        self.assertEqual(ex_data.level, "container")
        self.assertEqual(ex_data.num_devices, 1)
        self.assertEqual(ex_data.uuid, "b42756a2-37e4-3e47-674b-d1dd6e822145")
        self.assertEqual(ex_data.device, None)

    def test_no_metadata_examine(self):
        """Verify that md_examine works as expected with no metadata spec"""

        # shouldn't raise any exception
        with fake_utils("tests/mdadm_no_metadata_examine"):
            ex_data = BlockDev.md_examine("fake_dev")

        self.assertIs(ex_data.metadata, None)

    def test_fw_raid_migrating(self):
        """Verify that md_examine works when array is migrating ("foo <-- bar" values in output) """

        with fake_utils("tests/mdadm_fw_RAID_examine_migrate"):
            ex_data = BlockDev.md_examine("fake_dev")

        self.assertEqual(ex_data.chunk_size, 128 * 1024)

    def test_mdadm_name_extra_info(self):
        """Verify that md_examine and md_detail work with extra MD RAID name info"""

        with fake_utils("tests/mdadm_extra_name_stuff"):
            ex_data = BlockDev.md_examine("fake_dev")
            detail_data = BlockDev.md_detail("fake_dev")

        self.assertEqual(ex_data.name, "localhost:fedora")
        self.assertEqual(detail_data.name, "localhost:fedora")


class MDUnloadTest(MDTestCase):
    def setUp(self):
        # make sure the library is initialized with all plugins loaded for other
        # tests
        self.addCleanup(BlockDev.reinit, self.requested_plugins, True, None)

    def test_check_low_version(self):
        """Verify that checking the minimum mdsetup version works as expected"""

        # unload all plugins first
        self.assertTrue(BlockDev.reinit([], True, None))

        with fake_utils("tests/mdraid_low_version/"):
            # too low version of mdsetup available, the MD plugin should fail to load
            with self.assertRaises(GLib.GError):
                BlockDev.reinit(self.requested_plugins, True, None)

            self.assertNotIn("mdraid", BlockDev.get_available_plugin_names())

        # load the plugins back
        self.assertTrue(BlockDev.reinit(self.requested_plugins, True, None))
        self.assertIn("mdraid", BlockDev.get_available_plugin_names())

    def test_check_no_md(self):
        """Verify that checking mdsetup tool availability works as expected"""

        # unload all plugins first
        self.assertTrue(BlockDev.reinit([], True, None))

        with fake_path(all_but="mdadm"):
            # no mdadm available, the MD plugin should fail to load
            with self.assertRaises(GLib.GError):
                BlockDev.reinit(self.requested_plugins, True, None)

            self.assertNotIn("mdraid", BlockDev.get_available_plugin_names())

        # load the plugins back
        self.assertTrue(BlockDev.reinit(self.requested_plugins, True, None))
        self.assertIn("mdraid", BlockDev.get_available_plugin_names())