From 1688f554d23f2072f132777cbd55c583a31a41f6 Mon Sep 17 00:00:00 2001 From: Michael Still Date: Sun, 9 Dec 2012 21:00:09 +1100 Subject: Improve virt/disk/mount/nbd test coverage. Change-Id: I7d1c413b7a492c078ae10c840c4c20f8cf69d6a5 --- nova/tests/virt/disk/test_nbd.py | 203 ++++++++++++++++++++++++++++++++++++--- 1 file changed, 189 insertions(+), 14 deletions(-) (limited to 'nova') 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() -- cgit