| #!/usr/bin/python |
| # |
| |
| # Copyright (C) 2006, 2007, 2010, 2012, 2013, 2016 Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # 1. Redistributions of source code must retain the above copyright notice, |
| # this list of conditions and the following disclaimer. |
| # |
| # 2. Redistributions in binary form must reproduce the above copyright |
| # notice, this list of conditions and the following disclaimer in the |
| # documentation and/or other materials provided with the distribution. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS |
| # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED |
| # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR |
| # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR |
| # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, |
| # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, |
| # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR |
| # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF |
| # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING |
| # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS |
| # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| |
| """Script for unittesting the bdev module""" |
| |
| |
| import os |
| import random |
| import unittest |
| |
| from ganeti import compat |
| from ganeti import constants |
| from ganeti import errors |
| from ganeti import objects |
| from ganeti import utils |
| from ganeti.storage import bdev |
| |
| import testutils |
| |
| def _FakeRunCmd(success, stdout, cmd): |
| if success: |
| exit_code = 0 |
| else: |
| exit_code = 1 |
| return utils.RunResult(exit_code, None, stdout, "", cmd, |
| utils.process._TIMEOUT_NONE, 5) |
| |
| |
| class FakeStatResult(object): |
| def __init__(self, st_mode): |
| self.st_mode = st_mode |
| self.st_rdev = 0 |
| |
| |
| class TestRADOSBlockDevice(testutils.GanetiTestCase): |
| """Tests for bdev.RADOSBlockDevice volumes |
| |
| """ |
| |
| def setUp(self): |
| """Set up input data""" |
| testutils.GanetiTestCase.setUp(self) |
| |
| self.plain_output_old_ok = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_old_ok.txt") |
| self.plain_output_old_no_matches = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_old_no_matches.txt") |
| self.plain_output_old_extra_matches = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_old_extra_matches.txt") |
| self.plain_output_old_empty = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_old_empty.txt") |
| self.plain_output_new_ok = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_new_ok.txt") |
| self.plain_output_new_no_matches = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_new_no_matches.txt") |
| self.plain_output_new_extra_matches = \ |
| testutils.ReadTestData("bdev-rbd/plain_output_new_extra_matches.txt") |
| # This file is completely empty, and as such it's not shipped. |
| self.plain_output_new_empty = "" |
| self.json_output_ok = testutils.ReadTestData("bdev-rbd/json_output_ok.txt") |
| self.json_output_no_matches = \ |
| testutils.ReadTestData("bdev-rbd/json_output_no_matches.txt") |
| self.json_output_extra_matches = \ |
| testutils.ReadTestData("bdev-rbd/json_output_extra_matches.txt") |
| self.json_output_empty = \ |
| testutils.ReadTestData("bdev-rbd/json_output_empty.txt") |
| self.output_invalid = testutils.ReadTestData("bdev-rbd/output_invalid.txt") |
| |
| self.volume_name = "d7ab910a-4933-4ffe-88d0-faf2ce31390a.rbd.disk0" |
| self.test_unique_id = ("rbd", self.volume_name) |
| self.test_params = { |
| constants.LDP_POOL: "fake_pool" |
| } |
| |
| def testParseRbdShowmappedJson(self): |
| parse_function = bdev.RADOSBlockDevice._ParseRbdShowmappedJson |
| |
| self.assertEqual(parse_function(self.json_output_ok, self.volume_name), |
| "/dev/rbd3") |
| self.assertEqual(parse_function(self.json_output_empty, self.volume_name), |
| None) |
| self.assertEqual(parse_function(self.json_output_no_matches, |
| self.volume_name), None) |
| self.assertRaises(errors.BlockDeviceError, parse_function, |
| self.json_output_extra_matches, self.volume_name) |
| self.assertRaises(errors.BlockDeviceError, parse_function, |
| self.output_invalid, self.volume_name) |
| |
| def testParseRbdShowmappedPlain(self): |
| parse_function = bdev.RADOSBlockDevice._ParseRbdShowmappedPlain |
| |
| self.assertEqual(parse_function(self.plain_output_new_ok, |
| self.volume_name), "/dev/rbd3") |
| self.assertEqual(parse_function(self.plain_output_old_ok, |
| self.volume_name), "/dev/rbd3") |
| self.assertEqual(parse_function(self.plain_output_new_empty, |
| self.volume_name), None) |
| self.assertEqual(parse_function(self.plain_output_old_empty, |
| self.volume_name), None) |
| self.assertEqual(parse_function(self.plain_output_new_no_matches, |
| self.volume_name), None) |
| self.assertEqual(parse_function(self.plain_output_old_no_matches, |
| self.volume_name), None) |
| self.assertRaises(errors.BlockDeviceError, parse_function, |
| self.plain_output_new_extra_matches, self.volume_name) |
| self.assertRaises(errors.BlockDeviceError, parse_function, |
| self.plain_output_old_extra_matches, self.volume_name) |
| self.assertRaises(errors.BlockDeviceError, parse_function, |
| self.output_invalid, self.volume_name) |
| |
| @testutils.patch_object(utils, "RunCmd") |
| @testutils.patch_object(bdev.RADOSBlockDevice, "_UnmapVolumeFromBlockdev") |
| @testutils.patch_object(bdev.RADOSBlockDevice, "Attach") |
| def testRADOSBlockDeviceImport(self, attach_mock, unmap_mock, run_cmd_mock): |
| """Test for bdev.RADOSBlockDevice.Import()""" |
| # Set up the mock objects return values |
| attach_mock.return_value = True |
| run_cmd_mock.return_value = _FakeRunCmd(True, "", "") |
| |
| # Create a fake rbd volume |
| inst = bdev.RADOSBlockDevice(self.test_unique_id, [], 1024, |
| self.test_params, {}) |
| # Desired output command |
| import_cmd = [constants.RBD_CMD, "import", |
| "-p", inst.rbd_pool, |
| "-", inst.rbd_name] |
| |
| self.assertEqual(inst.Import(), import_cmd) |
| |
| @testutils.patch_object(bdev.RADOSBlockDevice, "Attach") |
| def testRADOSBlockDeviceExport(self, attach_mock): |
| """Test for bdev.RADOSBlockDevice.Export()""" |
| # Set up the mock object return value |
| attach_mock.return_value = True |
| |
| # Create a fake rbd volume |
| inst = bdev.RADOSBlockDevice(self.test_unique_id, [], 1024, |
| self.test_params, {}) |
| # Desired output command |
| export_cmd = [constants.RBD_CMD, "export", |
| "-p", inst.rbd_pool, |
| inst.rbd_name, "-"] |
| |
| self.assertEqual(inst.Export(), export_cmd) |
| |
| @testutils.patch_object(utils, "RunCmd") |
| @testutils.patch_object(bdev.RADOSBlockDevice, "Attach") |
| def testRADOSBlockDeviceCreate(self, attach_mock, run_cmd_mock): |
| """Test for bdev.RADOSBlockDevice.Create() success""" |
| attach_mock.return_value = True |
| # This returns a successful RunCmd result |
| run_cmd_mock.return_value = _FakeRunCmd(True, "", "") |
| |
| expect = bdev.RADOSBlockDevice(self.test_unique_id, [], 1024, |
| self.test_params, {}) |
| got = bdev.RADOSBlockDevice.Create(self.test_unique_id, [], 1024, None, |
| self.test_params, False, {}, |
| test_kwarg="test") |
| |
| self.assertEqual(expect, got) |
| |
| @testutils.patch_object(bdev.RADOSBlockDevice, "Attach") |
| def testRADOSBlockDeviceCreateFailure(self, attach_mock): |
| """Test for bdev.RADOSBlockDevice.Create() failure with exclusive_storage |
| enabled |
| |
| """ |
| attach_mock.return_value = True |
| |
| self.assertRaises(errors.ProgrammerError, bdev.RADOSBlockDevice.Create, |
| self.test_unique_id, [], 1024, None, self.test_params, |
| True, {}) |
| |
| @testutils.patch_object(bdev.RADOSBlockDevice, "_MapVolumeToBlockdev") |
| @testutils.patch_object(os, "stat") |
| def testAttach(self, stat_mock, map_mock): |
| """Test for bdev.RADOSBlockDevice.Attach()""" |
| stat_mock.return_value = FakeStatResult(0x6000) # bitmask for S_ISBLK |
| map_mock.return_value = "/fake/path" |
| dev = bdev.RADOSBlockDevice.__new__(bdev.RADOSBlockDevice) |
| dev.unique_id = self.test_unique_id |
| |
| self.assertEqual(dev.Attach(), True) |
| |
| @testutils.patch_object(bdev.RADOSBlockDevice, "_MapVolumeToBlockdev") |
| @testutils.patch_object(os, "stat") |
| def testAttachFailureNotBlockdev(self, stat_mock, map_mock): |
| """Test for bdev.RADOSBlockDevice.Attach() failure, not a blockdev""" |
| stat_mock.return_value = FakeStatResult(0x0) |
| map_mock.return_value = "/fake/path" |
| dev = bdev.RADOSBlockDevice.__new__(bdev.RADOSBlockDevice) |
| dev.unique_id = self.test_unique_id |
| |
| self.assertEqual(dev.Attach(), False) |
| |
| @testutils.patch_object(bdev.RADOSBlockDevice, "_MapVolumeToBlockdev") |
| @testutils.patch_object(os, "stat") |
| def testAttachFailureNoDevice(self, stat_mock, map_mock): |
| """Test for bdev.RADOSBlockDevice.Attach() failure, no device found""" |
| stat_mock.side_effect = OSError("No device found") |
| map_mock.return_value = "/fake/path" |
| dev = bdev.RADOSBlockDevice.__new__(bdev.RADOSBlockDevice) |
| dev.unique_id = self.test_unique_id |
| |
| self.assertEqual(dev.Attach(), False) |
| |
| |
| class TestExclusiveStoragePvs(unittest.TestCase): |
| """Test cases for functions dealing with LVM PV and exclusive storage""" |
| # Allowance for rounding |
| _EPS = 1e-4 |
| _MARGIN = constants.PART_MARGIN + constants.PART_RESERVED + _EPS |
| |
| @staticmethod |
| def _GenerateRandomPvInfo(rnd, name, vg): |
| # Granularity is .01 MiB |
| size = rnd.randint(1024 * 100, 10 * 1024 * 1024 * 100) |
| if rnd.choice([False, True]): |
| free = float(rnd.randint(0, size)) / 100.0 |
| else: |
| free = float(size) / 100.0 |
| size = float(size) / 100.0 |
| attr = "a-" |
| return objects.LvmPvInfo(name=name, vg_name=vg, size=size, free=free, |
| attributes=attr) |
| |
| def testGetStdPvSize(self): |
| """Test cases for bdev.LogicalVolume._GetStdPvSize()""" |
| rnd = random.Random(9517) |
| for _ in range(0, 50): |
| # Identical volumes |
| pvi = self._GenerateRandomPvInfo(rnd, "disk", "myvg") |
| onesize = bdev.LogicalVolume._GetStdPvSize([pvi]) |
| self.assertTrue(onesize <= pvi.size) |
| self.assertTrue(onesize > pvi.size * (1 - self._MARGIN)) |
| for length in range(2, 10): |
| n_size = bdev.LogicalVolume._GetStdPvSize([pvi] * length) |
| self.assertEqual(onesize, n_size) |
| |
| # Mixed volumes |
| for length in range(1, 10): |
| pvlist = [self._GenerateRandomPvInfo(rnd, "disk", "myvg") |
| for _ in range(0, length)] |
| std_size = bdev.LogicalVolume._GetStdPvSize(pvlist) |
| self.assertTrue(compat.all(std_size <= pvi.size for pvi in pvlist)) |
| self.assertTrue(compat.any(std_size > pvi.size * (1 - self._MARGIN) |
| for pvi in pvlist)) |
| pvlist.append(pvlist[0]) |
| p1_size = bdev.LogicalVolume._GetStdPvSize(pvlist) |
| self.assertEqual(std_size, p1_size) |
| |
| def testComputeNumPvs(self): |
| """Test cases for bdev.LogicalVolume._ComputeNumPvs()""" |
| rnd = random.Random(8067) |
| for _ in range(0, 1000): |
| pvlist = [self._GenerateRandomPvInfo(rnd, "disk", "myvg")] |
| lv_size = float(rnd.randint(10 * 100, 1024 * 1024 * 100)) / 100.0 |
| num_pv = bdev.LogicalVolume._ComputeNumPvs(lv_size, pvlist) |
| std_size = bdev.LogicalVolume._GetStdPvSize(pvlist) |
| self.assertTrue(num_pv >= 1) |
| self.assertTrue(num_pv * std_size >= lv_size) |
| self.assertTrue((num_pv - 1) * std_size < lv_size * (1 + self._EPS)) |
| |
| def testGetEmptyPvNames(self): |
| """Test cases for bdev.LogicalVolume._GetEmptyPvNames()""" |
| rnd = random.Random(21126) |
| for _ in range(0, 100): |
| num_pvs = rnd.randint(1, 20) |
| pvlist = [self._GenerateRandomPvInfo(rnd, "disk%d" % n, "myvg") |
| for n in range(0, num_pvs)] |
| for num_req in range(1, num_pvs + 2): |
| epvs = bdev.LogicalVolume._GetEmptyPvNames(pvlist, num_req) |
| epvs_set = compat.UniqueFrozenset(epvs) |
| if len(epvs) > 1: |
| self.assertEqual(len(epvs), len(epvs_set)) |
| for pvi in pvlist: |
| if pvi.name in epvs_set: |
| self.assertEqual(pvi.size, pvi.free) |
| else: |
| # There should be no remaining empty PV when less than the |
| # requeste number of PVs has been returned |
| self.assertTrue(len(epvs) == num_req or pvi.free != pvi.size) |
| |
| |
| class TestLogicalVolume(testutils.GanetiTestCase): |
| """Tests for bdev.LogicalVolume.""" |
| |
| def setUp(self): |
| """Set up test data""" |
| testutils.GanetiTestCase.setUp(self) |
| |
| self.volume_name = "31225655-5775-4356-c212-e8b1e137550a.disk0" |
| self.test_unique_id = ("ganeti", self.volume_name) |
| self.test_params = { |
| constants.LDP_STRIPES: 1 |
| } |
| self.pv_info_return = [objects.LvmPvInfo(name="/dev/sda5", vg_name="xenvg", |
| size=3500000.00, free=5000000.00, |
| attributes="wz--n-", lv_list=[])] |
| self.pv_info_invalid = [objects.LvmPvInfo(name="/dev/s:da5", |
| vg_name="xenvg", |
| size=3500000.00, free=5000000.00, |
| attributes="wz--n-", lv_list=[])] |
| self.pv_info_no_space = [objects.LvmPvInfo(name="/dev/sda5", vg_name="xenvg", |
| size=3500000.00, free=0.00, |
| attributes="wz--n-", lv_list=[])] |
| |
| |
| def testParseLvInfoLine(self): |
| """Tests for LogicalVolume._ParseLvInfoLine.""" |
| broken_lines = [ |
| " toomuch#vg#lv#-wi-ao#253#3#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#4096.00#/dev/abc(20)", |
| " vg#lv#-wi-a#253#3#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#25.3#3#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#twenty#3#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3.1#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#three#4096.00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#four#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#4096..00#2#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#4096.00#2.0#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#4096.00#two#/dev/abc(20)", |
| " vg#lv#-wi-ao#253#3#4096.00#2#/dev/abc20", |
| ] |
| for broken in broken_lines: |
| self.assertRaises(errors.BlockDeviceError, |
| bdev.LogicalVolume._ParseLvInfoLine, broken, "#") |
| |
| # Examples of good lines from "lvs": |
| # |
| # /dev/something|-wi-ao|253|3|4096.00|2|/dev/sdb(144),/dev/sdc(0) |
| # /dev/somethingelse|-wi-a-|253|4|4096.00|1|/dev/sdb(208) |
| true_out = [ |
| (("vg", "lv"), ("-wi-ao", 253, 3, 4096.00, 2, ["/dev/abc"])), |
| (("vg", "lv"), ("-wi-a-", 253, 7, 4096.00, 4, ["/dev/abc"])), |
| (("vg", "lv"), ("-ri-a-", 253, 4, 4.00, 5, ["/dev/abc", "/dev/def"])), |
| (("vg", "lv"), ("-wc-ao", 15, 18, 4096.00, 32, |
| ["/dev/abc", "/dev/def", "/dev/ghi0"])), |
| # Physical devices might be missing with thin volumes |
| (("vg", "lv"), ("twc-ao", 15, 18, 4096.00, 32, [])), |
| ] |
| for exp in true_out: |
| for sep in "#;|": |
| # NB We get lvs to return vg_name and lv_name separately, but |
| # _ParseLvInfoLine returns a pathname built from these, so we |
| # need to do some extra munging to round-trip this properly. |
| vg_name, lv_name = exp[0] |
| dev = os.environ.get('DM_DEV_DIR', '/dev') |
| devpath = os.path.join(dev, vg_name, lv_name) |
| lvs = exp[1] |
| pvs = ",".join("%s(%s)" % (d, i * 12) for (i, d) in enumerate(lvs[-1])) |
| fmt_str = sep.join((" %s", "%s", "%s", "%d", "%d", "%.2f", "%d", "%s")) |
| lvs_line = fmt_str % ((vg_name, lv_name) + lvs[0:-1] + (pvs,)) |
| parsed = bdev.LogicalVolume._ParseLvInfoLine(lvs_line, sep) |
| self.assertEqual(parsed, (devpath,) + exp[1:]) |
| |
| |
| def testGetLvGlobalInfo(self): |
| """Tests for LogicalVolume.GetLvGlobalInfo.""" |
| |
| good_lines="vg|1|-wi-ao|253|3|4096.00|2|/dev/sda(20)\n" \ |
| "vg|2|-wi-ao|253|3|4096.00|2|/dev/sda(21)\n" |
| expected_output = {"/dev/vg/1": ("-wi-ao", 253, 3, 4096, 2, ["/dev/sda"]), |
| "/dev/vg/2": ("-wi-ao", 253, 3, 4096, 2, ["/dev/sda"])} |
| |
| self.assertEqual({}, |
| bdev.LogicalVolume.GetLvGlobalInfo( |
| _run_cmd=lambda cmd: _FakeRunCmd(False, |
| "Fake error msg", |
| cmd))) |
| self.assertEqual({}, |
| bdev.LogicalVolume.GetLvGlobalInfo( |
| _run_cmd=lambda cmd: _FakeRunCmd(True, |
| "", |
| cmd))) |
| self.assertRaises(errors.BlockDeviceError, |
| bdev.LogicalVolume.GetLvGlobalInfo, |
| _run_cmd=lambda cmd: _FakeRunCmd(True, "BadStdOut", cmd)) |
| |
| fake_cmd = lambda cmd: _FakeRunCmd(True, good_lines, cmd) |
| good_res = bdev.LogicalVolume.GetLvGlobalInfo(_run_cmd=fake_cmd) |
| self.assertEqual(expected_output, good_res) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testLogicalVolumeImport(self, attach_mock): |
| """Tests for bdev.LogicalVolume.Import()""" |
| # Set up the mock object return value |
| attach_mock.return_value = True |
| |
| # Create a fake logical volume |
| inst = bdev.LogicalVolume(self.test_unique_id, [], 1024, {}, {}) |
| |
| # Desired output command |
| import_cmd = [constants.DD_CMD, |
| "of=%s" % inst.dev_path, |
| "bs=%s" % constants.DD_BLOCK_SIZE, |
| "oflag=direct", "conv=notrunc"] |
| |
| self.assertEqual(inst.Import(), import_cmd) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testLogicalVolumeExport(self, attach_mock): |
| """Test for bdev.LogicalVolume.Export()""" |
| # Set up the mock object return value |
| attach_mock.return_value = True |
| |
| # Create a fake logical volume |
| inst = bdev.LogicalVolume(self.test_unique_id, [], 1024, {}, {}) |
| |
| # Desired output command |
| export_cmd = [constants.DD_CMD, |
| "if=%s" % inst.dev_path, |
| "bs=%s" % constants.DD_BLOCK_SIZE, |
| "count=%s" % inst.size, |
| "iflag=direct"] |
| |
| self.assertEqual(inst.Export(), export_cmd) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(utils, "RunCmd") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreate(self, attach_mock, run_cmd_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() success""" |
| attach_mock.return_value = True |
| # This returns a successful RunCmd result |
| run_cmd_mock.return_value = _FakeRunCmd(True, "", "") |
| pv_info_mock.return_value = self.pv_info_return |
| |
| expect = bdev.LogicalVolume(self.test_unique_id, [], 1024, |
| self.test_params, {}) |
| got = bdev.LogicalVolume.Create(self.test_unique_id, [], 1024, None, |
| self.test_params, False, {}, |
| test_kwarg="test") |
| |
| self.assertEqual(expect, got) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailurePvsInfoExclStor(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when pv_info is empty and |
| exclusive storage is enabled |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = [] |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, {}, True, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailurePvsInfoNoExclStor(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when pv_info is empty and |
| exclusive storage is disabled |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = [] |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, {}, False, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailurePvsInvalid(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when pvs_info output is |
| invalid |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_invalid |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, {}, False, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailureNoSpindles(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when there are no spindles |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_return |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, |
| self.test_params,True, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailureNotEnoughSpindles(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when there are not enough |
| spindles |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_return |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, 0, |
| self.test_params, True, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailureNotEnoughEmptyPvs(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when there are not enough |
| empty pvs |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_return |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, 2, |
| self.test_params, True, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailureNoFreeSpace(self, attach_mock, pv_info_mock): |
| """Test for bdev.LogicalVolume.Create() failure when there is no free space |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_no_space |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, |
| self.test_params, False, {}) |
| |
| @testutils.patch_object(utils, "RunCmd") |
| @testutils.patch_object(bdev.LogicalVolume, "GetPVInfo") |
| @testutils.patch_object(bdev.LogicalVolume, "Attach") |
| def testCreateFailureCommand(self, attach_mock, pv_info_mock, run_cmd_mock): |
| """Test for bdev.LogicalVolume.Create() failure when the runcmd is incorrect |
| |
| """ |
| attach_mock.return_value = True |
| pv_info_mock.return_value = self.pv_info_return |
| run_cmd_mock = _FakeRunCmd(False, "", "") |
| |
| self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume.Create, |
| self.test_unique_id, [], 1024, None, |
| self.test_params, False, {}) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetLvGlobalInfo") |
| def testAttach(self, info_mock): |
| """Test for bdev.LogicalVolume.Attach()""" |
| info_mock.return_value = {"/dev/fake/path": ("v", 1, 0, 1024, 0, ["test"])} |
| dev = bdev.LogicalVolume.__new__(bdev.LogicalVolume) |
| dev.dev_path = "/dev/fake/path" |
| |
| self.assertEqual(dev.Attach(), True) |
| |
| @testutils.patch_object(bdev.LogicalVolume, "GetLvGlobalInfo") |
| def testAttachFalse(self, info_mock): |
| """Test for bdev.LogicalVolume.Attach() with missing lv_info""" |
| info_mock.return_value = {} |
| dev = bdev.LogicalVolume.__new__(bdev.LogicalVolume) |
| dev.dev_path = "/dev/fake/path" |
| |
| self.assertEqual(dev.Attach(), False) |
| |
| |
| class TestPersistentBlockDevice(testutils.GanetiTestCase): |
| """Tests for bdev.PersistentBlockDevice volumes |
| |
| """ |
| |
| def setUp(self): |
| """Set up test data""" |
| testutils.GanetiTestCase.setUp(self) |
| self.test_unique_id = (constants.BLOCKDEV_DRIVER_MANUAL, "/dev/abc") |
| |
| def testPersistentBlockDeviceImport(self): |
| """Test case for bdev.PersistentBlockDevice.Import()""" |
| # Create a fake block device |
| inst = bdev.PersistentBlockDevice(self.test_unique_id, [], 1024, {}, {}) |
| |
| self.assertRaises(errors.BlockDeviceError, |
| bdev.PersistentBlockDevice.Import, inst) |
| |
| @testutils.patch_object(bdev.PersistentBlockDevice, "Attach") |
| def testCreate(self, attach_mock): |
| """Test for bdev.PersistentBlockDevice.Create()""" |
| attach_mock.return_value = True |
| |
| expect = bdev.PersistentBlockDevice(self.test_unique_id, [], 0, {}, {}) |
| got = bdev.PersistentBlockDevice.Create(self.test_unique_id, [], 1024, None, |
| {}, False, {}, test_kwarg="test") |
| |
| self.assertEqual(expect, got) |
| |
| def testCreateFailure(self): |
| """Test for bdev.PersistentBlockDevice.Create() failure""" |
| |
| self.assertRaises(errors.ProgrammerError, bdev.PersistentBlockDevice.Create, |
| self.test_unique_id, [], 1024, None, {}, True, {}) |
| |
| @testutils.patch_object(os, "stat") |
| def testAttach(self, stat_mock): |
| """Test for bdev.PersistentBlockDevice.Attach()""" |
| stat_mock.return_value = FakeStatResult(0x6000) # bitmask for S_ISBLK |
| dev = bdev.PersistentBlockDevice.__new__(bdev.PersistentBlockDevice) |
| dev.dev_path = "/dev/fake/path" |
| |
| self.assertEqual(dev.Attach(), True) |
| |
| @testutils.patch_object(os, "stat") |
| def testAttachFailureNotBlockdev(self, stat_mock): |
| """Test for bdev.PersistentBlockDevice.Attach() failure, not a blockdev""" |
| stat_mock.return_value = FakeStatResult(0x0) |
| dev = bdev.PersistentBlockDevice.__new__(bdev.PersistentBlockDevice) |
| dev.dev_path = "/dev/fake/path" |
| |
| self.assertEqual(dev.Attach(), False) |
| |
| @testutils.patch_object(os, "stat") |
| def testAttachFailureNoDevice(self, stat_mock): |
| """Test for bdev.PersistentBlockDevice.Attach() failure, no device found""" |
| stat_mock.side_effect = OSError("No device found") |
| dev = bdev.PersistentBlockDevice.__new__(bdev.PersistentBlockDevice) |
| dev.dev_path = "/dev/fake/path" |
| |
| self.assertEqual(dev.Attach(), False) |
| |
| |
| if __name__ == "__main__": |
| testutils.GanetiTestProgram() |
