#!/usr/bin/python
#
-# Copyright (C) 2006, 2007 Google Inc.
+# Copyright (C) 2006, 2007, 2010 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
import unittest
from ganeti import bdev
+from ganeti import errors
+from ganeti import constants
+from ganeti import utils
+import testutils
-class TestDRBD8Runner(unittest.TestCase):
+
+class TestBaseDRBD(testutils.GanetiTestCase):
+ def testGetVersion(self):
+ data = [
+ ["version: 8.0.12 (api:76/proto:86-91)"],
+ ["version: 8.2.7 (api:88/proto:0-100)"],
+ ["version: 8.3.7.49 (api:188/proto:13-191)"],
+ ]
+ result = [
+ {
+ "k_major": 8,
+ "k_minor": 0,
+ "k_point": 12,
+ "api": 76,
+ "proto": 86,
+ "proto2": "91",
+ },
+ {
+ "k_major": 8,
+ "k_minor": 2,
+ "k_point": 7,
+ "api": 88,
+ "proto": 0,
+ "proto2": "100",
+ },
+ {
+ "k_major": 8,
+ "k_minor": 3,
+ "k_point": 7,
+ "api": 188,
+ "proto": 13,
+ "proto2": "191",
+ }
+ ]
+ for d,r in zip(data, result):
+ self.assertEqual(bdev.BaseDRBD._GetVersion(d), r)
+
+
+class TestDRBD8Runner(testutils.GanetiTestCase):
"""Testing case for DRBD8"""
@staticmethod
return retval
@staticmethod
- def _get_contents(name):
- """Returns the contents of a file"""
-
- prefix = os.environ.get("srcdir", None)
- if prefix:
- name = prefix + "/" + name
- fh = open(name, "r")
- try:
- data = fh.read()
- finally:
- fh.close()
- return data
-
-
- @staticmethod
def _has_net(data, local, remote):
"""Check network connection parameters"""
retval = (
"""Test drbdsetup show parser creation"""
bdev.DRBD8._GetShowParser()
- def testParserBoth(self):
- """Test drbdsetup show parser for disk and network"""
- data = self._get_contents("data/bdev-both.txt")
+ def testParser80(self):
+ """Test drbdsetup show parser for disk and network version 8.0"""
+ data = self._ReadTestData("bdev-drbd-8.0.txt")
result = bdev.DRBD8._GetDevInfo(data)
self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",
"/dev/xenvg/test.meta"),
"Wrong local disk info")
- self.failUnless(self._has_net(result, ("192.168.1.1", 11000),
- ("192.168.1.2", 11000)),
- "Wrong network info")
+ self.failUnless(self._has_net(result, ("192.0.2.1", 11000),
+ ("192.0.2.2", 11000)),
+ "Wrong network info (8.0.x)")
+
+ def testParser83(self):
+ """Test drbdsetup show parser for disk and network version 8.3"""
+ data = self._ReadTestData("bdev-drbd-8.3.txt")
+ result = bdev.DRBD8._GetDevInfo(data)
+ self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",
+ "/dev/xenvg/test.meta"),
+ "Wrong local disk info")
+ self.failUnless(self._has_net(result, ("192.0.2.1", 11000),
+ ("192.0.2.2", 11000)),
+ "Wrong network info (8.0.x)")
+
+ def testParserNetIP4(self):
+ """Test drbdsetup show parser for IPv4 network"""
+ data = self._ReadTestData("bdev-drbd-net-ip4.txt")
+ result = bdev.DRBD8._GetDevInfo(data)
+ self.failUnless(("local_dev" not in result and
+ "meta_dev" not in result and
+ "meta_index" not in result),
+ "Should not find local disk info")
+ self.failUnless(self._has_net(result, ("192.0.2.1", 11002),
+ ("192.0.2.2", 11002)),
+ "Wrong network info (IPv4)")
- def testParserNet(self):
- """Test drbdsetup show parser for disk and network"""
- data = self._get_contents("data/bdev-net.txt")
+ def testParserNetIP6(self):
+ """Test drbdsetup show parser for IPv6 network"""
+ data = self._ReadTestData("bdev-drbd-net-ip6.txt")
result = bdev.DRBD8._GetDevInfo(data)
self.failUnless(("local_dev" not in result and
"meta_dev" not in result and
"meta_index" not in result),
"Should not find local disk info")
- self.failUnless(self._has_net(result, ("192.168.1.1", 11002),
- ("192.168.1.2", 11002)),
- "Wrong network info")
+ self.failUnless(self._has_net(result, ("2001:db8:65::1", 11048),
+ ("2001:db8:66::1", 11048)),
+ "Wrong network info (IPv6)")
def testParserDisk(self):
- """Test drbdsetup show parser for disk and network"""
- data = self._get_contents("data/bdev-disk.txt")
+ """Test drbdsetup show parser for disk"""
+ data = self._ReadTestData("bdev-drbd-disk.txt")
result = bdev.DRBD8._GetDevInfo(data)
self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",
"/dev/xenvg/test.meta"),
"remote_addr" not in result),
"Should not find network info")
-if __name__ == '__main__':
- unittest.main()
+ def testBarriersOptions(self):
+ """Test class method that generates drbdsetup options for disk barriers"""
+ # Tests that should fail because of wrong version/options combinations
+ should_fail = [
+ (8, 0, 12, "bfd", True),
+ (8, 0, 12, "fd", False),
+ (8, 0, 12, "b", True),
+ (8, 2, 7, "bfd", True),
+ (8, 2, 7, "b", True)
+ ]
+
+ for vmaj, vmin, vrel, opts, meta in should_fail:
+ self.assertRaises(errors.BlockDeviceError,
+ bdev.DRBD8._ComputeDiskBarrierArgs,
+ vmaj, vmin, vrel, opts, meta)
+
+ # get the valid options from the frozenset(frozenset()) in constants.
+ valid_options = [list(x)[0] for x in constants.DRBD_VALID_BARRIER_OPT]
+
+ # Versions that do not support anything
+ for vmaj, vmin, vrel in ((8, 0, 0), (8, 0, 11), (8, 2, 6)):
+ for opts in valid_options:
+ self.assertRaises(errors.BlockDeviceError,
+ bdev.DRBD8._ComputeDiskBarrierArgs,
+ vmaj, vmin, vrel, opts, True)
+
+ # Versions with partial support (testing only options that are supported)
+ tests = [
+ (8, 0, 12, "n", False, []),
+ (8, 0, 12, "n", True, ["--no-md-flushes"]),
+ (8, 2, 7, "n", False, []),
+ (8, 2, 7, "fd", False, ["--no-disk-flushes", "--no-disk-drain"]),
+ (8, 0, 12, "n", True, ["--no-md-flushes"]),
+ ]
+
+ # Versions that support everything
+ for vmaj, vmin, vrel in ((8, 3, 0), (8, 3, 12)):
+ tests.append((vmaj, vmin, vrel, "bfd", True,
+ ["--no-disk-barrier", "--no-disk-drain",
+ "--no-disk-flushes", "--no-md-flushes"]))
+ tests.append((vmaj, vmin, vrel, "n", False, []))
+ tests.append((vmaj, vmin, vrel, "b", True,
+ ["--no-disk-barrier", "--no-md-flushes"]))
+ tests.append((vmaj, vmin, vrel, "fd", False,
+ ["--no-disk-flushes", "--no-disk-drain"]))
+ tests.append((vmaj, vmin, vrel, "n", True, ["--no-md-flushes"]))
+
+ # Test execution
+ for test in tests:
+ vmaj, vmin, vrel, disabled_barriers, disable_meta_flush, expected = test
+ args = \
+ bdev.DRBD8._ComputeDiskBarrierArgs(vmaj, vmin, vrel,
+ disabled_barriers,
+ disable_meta_flush)
+ self.failUnless(set(args) == set(expected),
+ "For test %s, got wrong results %s" % (test, args))
+
+ # Unsupported or invalid versions
+ for vmaj, vmin, vrel in ((0, 7, 25), (9, 0, 0), (7, 0, 0), (8, 4, 0)):
+ self.assertRaises(errors.BlockDeviceError,
+ bdev.DRBD8._ComputeDiskBarrierArgs,
+ vmaj, vmin, vrel, "n", True)
+
+ # Invalid options
+ for option in ("", "c", "whatever", "nbdfc", "nf"):
+ self.assertRaises(errors.BlockDeviceError,
+ bdev.DRBD8._ComputeDiskBarrierArgs,
+ 8, 3, 11, option, True)
+
+
+class TestDRBD8Status(testutils.GanetiTestCase):
+ """Testing case for DRBD8 /proc status"""
+
+ def setUp(self):
+ """Read in txt data"""
+ testutils.GanetiTestCase.setUp(self)
+ proc_data = self._TestDataFilename("proc_drbd8.txt")
+ proc80e_data = self._TestDataFilename("proc_drbd80-emptyline.txt")
+ proc83_data = self._TestDataFilename("proc_drbd83.txt")
+ proc83_sync_data = self._TestDataFilename("proc_drbd83_sync.txt")
+ proc83_sync_krnl_data = \
+ self._TestDataFilename("proc_drbd83_sync_krnl2.6.39.txt")
+ self.proc_data = bdev.DRBD8._GetProcData(filename=proc_data)
+ self.proc80e_data = bdev.DRBD8._GetProcData(filename=proc80e_data)
+ self.proc83_data = bdev.DRBD8._GetProcData(filename=proc83_data)
+ self.proc83_sync_data = bdev.DRBD8._GetProcData(filename=proc83_sync_data)
+ self.proc83_sync_krnl_data = \
+ bdev.DRBD8._GetProcData(filename=proc83_sync_krnl_data)
+ self.mass_data = bdev.DRBD8._MassageProcData(self.proc_data)
+ self.mass80e_data = bdev.DRBD8._MassageProcData(self.proc80e_data)
+ self.mass83_data = bdev.DRBD8._MassageProcData(self.proc83_data)
+ self.mass83_sync_data = bdev.DRBD8._MassageProcData(self.proc83_sync_data)
+ self.mass83_sync_krnl_data = \
+ bdev.DRBD8._MassageProcData(self.proc83_sync_krnl_data)
+
+ def testIOErrors(self):
+ """Test handling of errors while reading the proc file."""
+ temp_file = self._CreateTempFile()
+ os.unlink(temp_file)
+ self.failUnlessRaises(errors.BlockDeviceError,
+ bdev.DRBD8._GetProcData, filename=temp_file)
+
+ def testHelper(self):
+ """Test reading usermode_helper in /sys."""
+ sys_drbd_helper = self._TestDataFilename("sys_drbd_usermode_helper.txt")
+ drbd_helper = bdev.DRBD8.GetUsermodeHelper(filename=sys_drbd_helper)
+ self.failUnlessEqual(drbd_helper, "/bin/true")
+
+ def testHelperIOErrors(self):
+ """Test handling of errors while reading usermode_helper in /sys."""
+ temp_file = self._CreateTempFile()
+ os.unlink(temp_file)
+ self.failUnlessRaises(errors.BlockDeviceError,
+ bdev.DRBD8.GetUsermodeHelper, filename=temp_file)
+
+ def testMinorNotFound(self):
+ """Test not-found-minor in /proc"""
+ self.failUnless(9 not in self.mass_data)
+ self.failUnless(9 not in self.mass83_data)
+ self.failUnless(3 not in self.mass80e_data)
+
+ def testLineNotMatch(self):
+ """Test wrong line passed to DRBD8Status"""
+ self.assertRaises(errors.BlockDeviceError, bdev.DRBD8Status, "foo")
+
+ def testMinor0(self):
+ """Test connected, primary device"""
+ for data in [self.mass_data, self.mass83_data]:
+ stats = bdev.DRBD8Status(data[0])
+ self.failUnless(stats.is_in_use)
+ self.failUnless(stats.is_connected and stats.is_primary and
+ stats.peer_secondary and stats.is_disk_uptodate)
+
+ def testMinor1(self):
+ """Test connected, secondary device"""
+ for data in [self.mass_data, self.mass83_data]:
+ stats = bdev.DRBD8Status(data[1])
+ self.failUnless(stats.is_in_use)
+ self.failUnless(stats.is_connected and stats.is_secondary and
+ stats.peer_primary and stats.is_disk_uptodate)
+
+ def testMinor2(self):
+ """Test unconfigured device"""
+ for data in [self.mass_data, self.mass83_data, self.mass80e_data]:
+ stats = bdev.DRBD8Status(data[2])
+ self.failIf(stats.is_in_use)
+
+ def testMinor4(self):
+ """Test WFconn device"""
+ for data in [self.mass_data, self.mass83_data]:
+ stats = bdev.DRBD8Status(data[4])
+ self.failUnless(stats.is_in_use)
+ self.failUnless(stats.is_wfconn and stats.is_primary and
+ stats.rrole == 'Unknown' and
+ stats.is_disk_uptodate)
+
+ def testMinor6(self):
+ """Test diskless device"""
+ for data in [self.mass_data, self.mass83_data]:
+ stats = bdev.DRBD8Status(data[6])
+ self.failUnless(stats.is_in_use)
+ self.failUnless(stats.is_connected and stats.is_secondary and
+ stats.peer_primary and stats.is_diskless)
+
+ def testMinor8(self):
+ """Test standalone device"""
+ for data in [self.mass_data, self.mass83_data]:
+ stats = bdev.DRBD8Status(data[8])
+ self.failUnless(stats.is_in_use)
+ self.failUnless(stats.is_standalone and
+ stats.rrole == 'Unknown' and
+ stats.is_disk_uptodate)
+
+ def testDRBD83SyncFine(self):
+ stats = bdev.DRBD8Status(self.mass83_sync_data[3])
+ self.failUnless(stats.is_in_resync)
+ self.failUnless(stats.sync_percent is not None)
+
+ def testDRBD83SyncBroken(self):
+ stats = bdev.DRBD8Status(self.mass83_sync_krnl_data[3])
+ self.failUnless(stats.is_in_resync)
+ self.failUnless(stats.sync_percent is not None)
+
+
+class TestRADOSBlockDevice(testutils.GanetiTestCase):
+ def test_ParseRbdShowmappedOutput(self):
+ volume_name = "abc9778-8e8ace5b.rbd.disk0"
+ output_ok = \
+ ("0\trbd\te69f28e5-9817.rbd.disk0\t-\t/dev/rbd0\n"
+ "1\t/dev/rbd0\tabc9778-8e8ace5b.rbd.disk0\t-\t/dev/rbd16\n"
+ "line\twith\tfewer\tfields\n"
+ "")
+ output_empty = ""
+ output_no_matches = \
+ ("0\trbd\te69f28e5-9817.rbd.disk0\t-\t/dev/rbd0\n"
+ "1\trbd\tabcdef01-9817.rbd.disk0\t-\t/dev/rbd10\n"
+ "2\trbd\tcdef0123-9817.rbd.disk0\t-\t/dev/rbd12\n"
+ "something\twith\tfewer\tfields"
+ "")
+ output_extra_matches = \
+ ("0\t/dev/rbd0\tabc9778-8e8ace5b.rbd.disk0\t-\t/dev/rbd11\n"
+ "1\trbd\te69f28e5-9817.rbd.disk0\t-\t/dev/rbd0\n"
+ "2\t/dev/rbd0\tabc9778-8e8ace5b.rbd.disk0\t-\t/dev/rbd16\n"
+ "something\twith\tfewer\tfields"
+ "")
+
+ parse_function = bdev.RADOSBlockDevice._ParseRbdShowmappedOutput
+ self.assertEqual(parse_function(output_ok, volume_name), "/dev/rbd16")
+ self.assertRaises(errors.BlockDeviceError, parse_function,
+ output_empty, volume_name)
+ self.assertEqual(parse_function(output_no_matches, volume_name), None)
+ self.assertRaises(errors.BlockDeviceError, parse_function,
+ output_extra_matches, volume_name)
+
+
+class TestCheckFileStoragePath(unittest.TestCase):
+ def testNonAbsolute(self):
+ for i in ["", "tmp", "foo/bar/baz"]:
+ self.assertRaises(errors.FileStoragePathError,
+ bdev._CheckFileStoragePath, i, ["/tmp"])
+
+ self.assertRaises(errors.FileStoragePathError,
+ bdev._CheckFileStoragePath, "/tmp", ["tmp", "xyz"])
+
+ def testNoAllowed(self):
+ self.assertRaises(errors.FileStoragePathError,
+ bdev._CheckFileStoragePath, "/tmp", [])
+
+ def testNoAdditionalPathComponent(self):
+ self.assertRaises(errors.FileStoragePathError,
+ bdev._CheckFileStoragePath, "/tmp/foo", ["/tmp/foo"])
+
+ def testAllowed(self):
+ bdev._CheckFileStoragePath("/tmp/foo/a", ["/tmp/foo"])
+ bdev._CheckFileStoragePath("/tmp/foo/a/x", ["/tmp/foo"])
+
+
+class TestLoadAllowedFileStoragePaths(testutils.GanetiTestCase):
+ def testDevNull(self):
+ self.assertEqual(bdev.LoadAllowedFileStoragePaths("/dev/null"), [])
+
+ def testNonExistantFile(self):
+ filename = "/tmp/this/file/does/not/exist"
+ assert not os.path.exists(filename)
+ self.assertEqual(bdev.LoadAllowedFileStoragePaths(filename), [])
+
+ def test(self):
+ tmpfile = self._CreateTempFile()
+
+ utils.WriteFile(tmpfile, data="""
+ # This is a test file
+ /tmp
+ /srv/storage
+ relative/path
+ """)
+
+ self.assertEqual(bdev.LoadAllowedFileStoragePaths(tmpfile), [
+ "/tmp",
+ "/srv/storage",
+ "relative/path",
+ ])
+
+
+if __name__ == "__main__":
+ testutils.GanetiTestProgram()
projects / ganeti-local / blobdiff