[master 11/21] Use availability information in device actions.
mulkieran
installerbot-noreply at redhat.com
Tue May 12 21:35:53 UTC 2015
From: mulhern <amulhern at redhat.com>
Related: #12
Signed-off-by: mulhern <amulhern at redhat.com>
---
blivet/deviceaction.py | 12 +++++
tests/devices_test/dependencies_test.py | 90 +++++++++++++++++++++++++++++++++
2 files changed, 102 insertions(+)
create mode 100644 tests/devices_test/dependencies_test.py
diff --git a/blivet/deviceaction.py b/blivet/deviceaction.py
index d38f99c..e6a0e7c 100644
--- a/blivet/deviceaction.py
+++ b/blivet/deviceaction.py
@@ -148,6 +148,12 @@ 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:
+ dependencies_str = ", ".join(str(d) for d in unavailable_dependencies)
+ raise ValueError("device type %s requires unavailable_dependencies: %s" % (device.type, dependencies_str))
+
self.device = device
self.container = getattr(self.device, "container", None)
self._applied = False
@@ -520,6 +526,9 @@ def __init__(self, device, fmt=None):
if self._format.exists:
raise ValueError("specified format already exists")
+ if not self._format.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:
@@ -642,6 +651,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..55ea3de
--- /dev/null
+++ b/tests/devices_test/dependencies_test.py
@@ -0,0 +1,90 @@
+#!/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 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.assertGreater(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
+ self.dm_method = availability.BLOCKDEV_DM_PLUGIN._method
+ self.cache_availability = availability.CACHE_AVAILABILITY
+
+ def testAvailabilityMDRAIDplugin(self):
+
+ availability.CACHE_AVAILABILITY = False
+ availability.BLOCKDEV_DM_PLUGIN._method = availability.AvailableMethod
+
+ # 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 = availability.AvailableMethod
+ 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 = availability.UnavailableMethod
+ 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
+ availability.BLOCKDEV_DM_PLUGIN._method = self.dm_method
+
+ availability.CACHE_AVAILABILITY = False
+ availability.BLOCKDEV_MDRAID_PLUGIN.available # pylint: disable=pointless-statement
+ availability.BLOCKDEV_DM_PLUGIN.available # pylint: disable=pointless-statement
+
+ availability.CACHE_AVAILABILITY = self.cache_availability
+
+if __name__ == "__main__":
+ unittest.main()
--
To view this commit on github, visit https://github.com/rhinstaller/blivet/commit/8521b44d0a5e62997901bb34a9bca454d3edb0cc
More information about the anaconda-patches
mailing list