From: mulhern amulhern@redhat.com
Related: #12
Signed-off-by: mulhern amulhern@redhat.com --- blivet/deviceaction.py | 11 +++++ tests/devices_test/dependencies_test.py | 86 +++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+) create mode 100644 tests/devices_test/dependencies_test.py
diff --git a/blivet/deviceaction.py b/blivet/deviceaction.py index 4d10765..212dee3 100644 --- a/blivet/deviceaction.py +++ b/blivet/deviceaction.py @@ -148,6 +148,11 @@ def __init__(self, device): util.ObjectID.__init__(self) if not isinstance(device, StorageDevice): raise ValueError("arg 1 must be a StorageDevice instance") + + unavailable_dependencies = device.unavailableDependencies + if unavailable_dependencies: + raise ValueError("device type %s requires unavailable_dependencies: %s" % (device.type, ", ".join(str(d) for d in unavailable_dependencies))) + self.device = device self.container = getattr(self.device, "container", None) self._applied = False @@ -516,6 +521,9 @@ def __init__(self, device, fmt=None): if self._format.exists: raise ValueError("specified format already exists")
+ if not fmt.formattable: + raise ValueError("resource to create this format %s is unavailable" % fmt) + def apply(self): """ apply changes related to the action to the device(s) """ if self._applied: @@ -638,6 +646,9 @@ def __init__(self, device): DeviceAction.__init__(self, device) self.origFormat = self.device.format
+ if not device.format.destroyable: + raise ValueError("resource to destroy this format type %s is unavailable" % device.format.type) + def apply(self): if self._applied: return diff --git a/tests/devices_test/dependencies_test.py b/tests/devices_test/dependencies_test.py new file mode 100644 index 0000000..2077216 --- /dev/null +++ b/tests/devices_test/dependencies_test.py @@ -0,0 +1,86 @@ +#!/usr/bin/python +# vim:set fileencoding=utf-8 + +import unittest + +from blivet.deviceaction import ActionCreateDevice +from blivet.deviceaction import ActionDestroyDevice + +from blivet.deviceaction import ActionCreateFormat +from blivet.deviceaction import ActionDestroyFormat + +from blivet.devices import DiskDevice +from blivet.devices import LUKSDevice +from blivet.devices import MDRaidArrayDevice +from blivet.devices import PartitionDevice + +from blivet.formats import getFormat + +from blivet.tasks import availability + +class FalseMethod(availability.Method): + def available(self, resource): + return False + +class TrueMethod(availability.Method): + def available(self, resource): + return True + +class DeviceDependenciesTestCase(unittest.TestCase): + """Test external device dependencies. """ + + def testDependencies(self): + dev1 = DiskDevice("name", fmt=getFormat("mdmember")) + dev2 = DiskDevice("other", fmt=getFormat("mdmember")) + dev = MDRaidArrayDevice("dev", level="raid1", parents=[dev1,dev2]) + luks = LUKSDevice("luks", parents=[dev]) + + # a parent's dependencies are a subset of its child's. + for d in dev.externalDependencies: + self.assertIn(d, luks.externalDependencies) + + # make sure that there's at least something in these dependencies + self.assertTrue(len(luks.externalDependencies) > 0) + +class MockingDeviceDependenciesTestCase(unittest.TestCase): + """Test availability of external device dependencies. """ + + def setUp(self): + dev1 = DiskDevice("name", fmt=getFormat("mdmember")) + dev2 = DiskDevice("other") + self.part = PartitionDevice("part", fmt=getFormat("mdmember"), parents=[dev2]) + self.dev = MDRaidArrayDevice("dev", level="raid1", parents=[dev1, self.part], fmt=getFormat("luks")) + self.luks = LUKSDevice("luks", parents=[self.dev], fmt=getFormat("ext4")) + + self.mdraid_method = availability.BLOCKDEV_MDRAID_PLUGIN._method + + def testAvailabilityMDRAIDplugin(self): + + # if the plugin is not in, there's nothing to test + self.assertIn(availability.BLOCKDEV_MDRAID_PLUGIN, self.luks.externalDependencies) + + # dev is not among its unavailable dependencies + availability.BLOCKDEV_MDRAID_PLUGIN._method = TrueMethod() + self.assertNotIn(availability.BLOCKDEV_MDRAID_PLUGIN, self.luks.unavailableDependencies) + self.assertIsNotNone(ActionCreateDevice(self.luks)) + self.assertIsNotNone(ActionDestroyDevice(self.luks)) + self.assertIsNotNone(ActionCreateFormat(self.luks, fmt=getFormat("macefi"))) + self.assertIsNotNone(ActionDestroyFormat(self.luks)) + + # dev is among the unavailable dependencies + availability.BLOCKDEV_MDRAID_PLUGIN._method = FalseMethod() + self.assertIn(availability.BLOCKDEV_MDRAID_PLUGIN, self.luks.unavailableDependencies) + with self.assertRaises(ValueError): + ActionCreateDevice(self.luks) + with self.assertRaises(ValueError): + ActionDestroyDevice(self.dev) + with self.assertRaises(ValueError): + ActionCreateFormat(self.dev) + with self.assertRaises(ValueError): + ActionDestroyFormat(self.dev) + + def tearDown(self): + availability.BLOCKDEV_MDRAID_PLUGIN._method = self.mdraid_method + +if __name__ == "__main__": + unittest.main()