Add support for listing unmounted devices

Change-Id: Ia30b3104086a1b1bd45e14f7d328ce383d1e4ee0

ceph/__init__.py

@@ -23,6 +23,7 @@ import re
 import sys
 import errno
 import shutil
+import pyudev
 
 from charmhelpers.core import hookenv
 from charmhelpers.core.host import (
@@ -58,7 +59,7 @@ PEON = 'peon'
 QUORUM = [LEADER, PEON]
 
 PACKAGES = ['ceph', 'gdisk', 'ntp', 'btrfs-tools', 'python-ceph',
-            'radosgw', 'xfsprogs']
+            'radosgw', 'xfsprogs', 'python-pyudev']
 
 LinkSpeed = {
     "BASE_10": 10,
@@ -102,6 +103,23 @@ NETWORK_ADAPTER_SYSCTLS = {
 }
 
 
+def unmounted_disks():
+    """List of unmounted block devices on the current host."""
+    disks = []
+    context = pyudev.Context()
+    for device in context.list_devices(DEVTYPE='disk'):
+        if device['SUBSYSTEM'] == 'block':
+            matched = False
+            for block_type in [u'dm', u'loop', u'ram', u'nbd']:
+                if block_type in device.device_node:
+                    matched = True
+            if matched:
+                continue
+            disks.append(device.device_node)
+    log("Found disks: {}".format(disks))
+    return [disk for disk in disks if not is_device_mounted(disk)]
+
+
 def save_sysctls(sysctl_dict, save_location):
     """
     Persist the sysctls to the hard drive.

requirements.txt

@@ -3,3 +3,4 @@ nose
 # of appearance. Changing the order has an impact on the overall integration
 # process, which may cause wedges in the gate later.
 charmhelpers
+pyudev

unit_tests/test_ceph.py

@@ -14,10 +14,25 @@
 
 import mock
 import unittest
-
 import ceph
 
 
+class TestDevice():
+    """Test class to mock out pyudev Device"""
+
+    def __getitem__(**kwargs):
+        """
+        Mock [].
+
+        We need this method to be present in the test class mock even
+        though we mock the return value with the MagicMock later
+        """
+        return "Some device type"
+
+    def device_node():
+        "/dev/test_device"
+
+
 class CephTestCase(unittest.TestCase):
     def setUp(self):
         super(CephTestCase, self).setUp()
@@ -55,6 +70,31 @@ class CephTestCase(unittest.TestCase):
                          'osd',
                          'allow rwx'])
 
+    def test_list_unmounted_devices(self):
+        dev1 = mock.MagicMock(spec=TestDevice)
+        dev1.__getitem__.return_value = "block"
+        dev1.device_node = '/dev/sda'
+        dev2 = mock.MagicMock(spec=TestDevice)
+        dev2.__getitem__.return_value = "block"
+        dev2.device_node = '/dev/sdb'
+        dev3 = mock.MagicMock(spec=TestDevice)
+        dev3.__getitem__.return_value = "block"
+        dev3.device_node = '/dev/loop1'
+        devices = [dev1, dev2, dev3]
+        with mock.patch(
+            'pyudev.Context.list_devices',
+                return_value=devices):
+                with mock.patch.object(ceph,
+                                       'is_device_mounted',
+                                       return_value=False):
+                    devices = ceph.unmounted_disks()
+                    self.assertEqual(devices, ['/dev/sda', '/dev/sdb'])
+                with mock.patch.object(ceph,
+                                       'is_device_mounted',
+                                       return_value=True):
+                    devices = ceph.unmounted_disks()
+                    self.assertEqual(devices, [])
+
 
 class CephVersionTestCase(unittest.TestCase):