Merge "Improve virt/disk/mount/nbd test coverage."

1 files changed, 189 insertions, 14 deletions

diff --git a/nova/tests/virt/disk/test_nbd.py b/nova/tests/virt/disk/test_nbd.py
index b78378c95..59f1c29d8 100644
--- a/nova/tests/virt/disk/test_nbd.py
+++ b/nova/tests/virt/disk/test_nbd.py
@@ -18,15 +18,28 @@
 
 import fixtures
 import os
+from testtools.matchers import Equals
+from testtools.matchers import MatchesListwise
 
 from nova import test
 
 from nova.openstack.common import cfg
+from nova import utils
 from nova.virt.disk.mount import nbd
 
 CONF = cfg.CONF
 CONF.import_opt('max_nbd_devices', 'nova.virt.disk.mount.nbd')
 
+ORIG_EXISTS = os.path.exists
+
+
+def _fake_exists_no_users(path):
+    if path.startswith('/sys/block/nbd'):
+        if path.endswith('pid'):
+            return False
+        return True
+    return ORIG_EXISTS(path)
+
 
 class NbdTestCase(test.TestCase):
     def setUp(self):
@@ -39,10 +52,17 @@ class NbdTestCase(test.TestCase):
         tempdir = self.useFixture(fixtures.TempDir()).path
         n = nbd.NbdMount(None, tempdir)
         self.assertTrue(n._DEVICES_INITIALIZED)
-        self.assertEquals(['/dev/nbd0', '/dev/nbd1'], nbd.NbdMount._DEVICES)
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1')]))
+
+    def test_nbd_no_free_devices(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        nbd.NbdMount._DEVICES = []
+        self.assertEquals(None, n._allocate_nbd())
 
     def test_nbd_not_loaded(self):
-        orig_exists = os.path.exists
         tempdir = self.useFixture(fixtures.TempDir()).path
         n = nbd.NbdMount(None, tempdir)
 
@@ -50,7 +70,7 @@ class NbdTestCase(test.TestCase):
         def fake_exists(path):
             if path.startswith('/sys/block/nbd'):
                 return False
-            return orig_exists(path)
+            return ORIG_EXISTS(path)
         self.useFixture(fixtures.MonkeyPatch('os.path.exists', fake_exists))
 
         # This should fail, as we don't have the module "loaded"
@@ -60,28 +80,183 @@ class NbdTestCase(test.TestCase):
         self.assertEquals('nbd unavailable: module not loaded', n.error)
 
         # And no device should be consumed
-        self.assertEquals(['/dev/nbd0', '/dev/nbd1'], nbd.NbdMount._DEVICES)
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1')]))
+
+    def test_nbd_all_allocated(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        nbd.NbdMount._DEVICES = []
+        self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                             _fake_exists_no_users))
+
+        # Allocate a nbd device, fails
+        self.assertEquals(None, n._allocate_nbd())
+        self.assertEquals('No free nbd devices', n.error)
 
     def test_nbd_allocation(self):
-        orig_exists = os.path.exists
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                             _fake_exists_no_users))
+
+        # Allocate a nbd device
+        self.assertEquals('/dev/nbd1', n._allocate_nbd())
+        self.assertEquals(['/dev/nbd0'], nbd.NbdMount._DEVICES)
+
+        # Allocate another
+        self.assertEquals('/dev/nbd0', n._allocate_nbd())
+        self.assertEquals([], nbd.NbdMount._DEVICES)
+
+    def test_nbd_allocation_one_in_use(self):
         tempdir = self.useFixture(fixtures.TempDir()).path
         n = nbd.NbdMount(None, tempdir)
 
         # Fake out os.path.exists
         def fake_exists(path):
             if path.startswith('/sys/block/nbd'):
+                if path == '/sys/block/nbd1/pid':
+                    return True
                 if path.endswith('pid'):
                     return False
                 return True
-            return orig_exists(path)
+            return ORIG_EXISTS(path)
         self.useFixture(fixtures.MonkeyPatch('os.path.exists', fake_exists))
 
-        # Allocate a nbd device
-        d = n._allocate_nbd()
-        self.assertEquals('/dev/nbd1', d)
-        self.assertEquals(1, len(nbd.NbdMount._DEVICES))
-
-        # Allocate another
-        d = n._allocate_nbd()
-        self.assertEquals('/dev/nbd0', d)
+        # Allocate a nbd device, should not be the in use one
+        # TODO(mikal): Note that there is a leak here, as the in use nbd device
+        # is removed from the list, but not returned so it will never be
+        # re-added. I will fix this in a later patch.
+        self.assertEquals('/dev/nbd0', n._allocate_nbd())
         self.assertEquals([], nbd.NbdMount._DEVICES)
+        self.assertTrue('/dev/nbd0' not in nbd.NbdMount._DEVICES)
+
+    def test_free(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        d = '/dev/nosuch'
+        self.assertFalse(d in nbd.NbdMount._DEVICES)
+
+        # Now free it
+        n._free_nbd(d)
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1'),
+                                         Equals('/dev/nosuch')]))
+        self.assertTrue(d in nbd.NbdMount._DEVICES)
+
+        # Double free
+        n._free_nbd(d)
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1'),
+                                         Equals('/dev/nosuch')]))
+        self.assertTrue(d in nbd.NbdMount._DEVICES)
+
+    def test_get_dev_no_devices(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        nbd.NbdMount._DEVICES = []
+        self.assertFalse(n.get_dev())
+
+    def test_get_dev_qemu_fails(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                             _fake_exists_no_users))
+
+        # We have a trycmd that always fails
+        def fake_trycmd(*args, **kwargs):
+            return '', 'broken'
+        self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', fake_trycmd))
+
+        # Error logged, no device consumed
+        self.assertFalse(n.get_dev())
+        self.assertTrue(n.error.startswith('qemu-nbd error'))
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1')]))
+
+    def test_get_dev_qemu_timeout(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                             _fake_exists_no_users))
+
+        # We have a trycmd that always passed
+        def fake_trycmd(*args, **kwargs):
+            return '', ''
+        self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', fake_trycmd))
+
+        # We steal time.sleep() as well to speed up this test
+        def fake_sleep(duration):
+            return
+        self.useFixture(fixtures.MonkeyPatch('time.sleep', fake_sleep))
+
+        # Error logged, no device consumed
+        self.assertFalse(n.get_dev())
+        self.assertTrue(n.error.endswith('did not show up'))
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1')]))
+
+    def test_get_dev_works(self):
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+
+        # We need the pid file for the device which is allocated to exist, but
+        # only once it is allocated to us
+        def fake_exists_one(path):
+            if path.startswith('/sys/block/nbd'):
+                if path == '/sys/block/nbd1/pid':
+                    return False
+                if path.endswith('pid'):
+                    return False
+                return True
+            return ORIG_EXISTS(path)
+        self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                             fake_exists_one))
+
+        # We have a trycmd that always passed
+        def fake_trycmd(*args, **kwargs):
+            def fake_exists_two(path):
+                if path.startswith('/sys/block/nbd'):
+                    if path == '/sys/block/nbd1/pid':
+                        return True
+                    if path.endswith('pid'):
+                        return False
+                    return True
+                return ORIG_EXISTS(path)
+            self.useFixture(fixtures.MonkeyPatch('os.path.exists',
+                                                 fake_exists_two))
+            return '', ''
+        self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', fake_trycmd))
+
+        # Ditto execute
+        def fake_exec(*args, **kwargs):
+            return
+        self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', fake_exec))
+
+        # No error logged, device consumed
+        self.assertTrue(n.get_dev())
+        self.assertTrue(n.linked)
+        self.assertEquals('', n.error)
+        self.assertEquals(['/dev/nbd0'], nbd.NbdMount._DEVICES)
+        self.assertEquals('/dev/nbd1', n.device)
+
+        # Free
+        n.unget_dev()
+        self.assertFalse(n.linked)
+        self.assertEquals('', n.error)
+        self.assertThat(nbd.NbdMount._DEVICES,
+                        MatchesListwise([Equals('/dev/nbd0'),
+                                         Equals('/dev/nbd1')]))
+        self.assertEquals(None, n.device)
+
+    def test_unget_dev_simple(self):
+        # This test is just checking we don't get an exception when we unget
+        # something we don't have
+        tempdir = self.useFixture(fixtures.TempDir()).path
+        n = nbd.NbdMount(None, tempdir)
+        n.unget_dev()