"""Script for unittesting the bdev module"""
+import os
import unittest
from ganeti import bdev
+from ganeti import errors
class TestDRBD8Runner(unittest.TestCase):
return retval
@staticmethod
+ def _get_contents(name):
+ """Returns the contents of a file"""
+
+ prefix = os.environ.get("srcdir", None)
+ if prefix:
+ name = prefix + "/test/" + 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 = (
def testParserBoth(self):
"""Test drbdsetup show parser for disk and network"""
- data = open("data/bdev-both.txt").read()
+ data = self._get_contents("data/bdev-both.txt")
result = bdev.DRBD8._GetDevInfo(data)
self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",
"/dev/xenvg/test.meta"),
def testParserNet(self):
"""Test drbdsetup show parser for disk and network"""
- data = open("data/bdev-net.txt").read()
+ data = self._get_contents("data/bdev-net.txt")
result = bdev.DRBD8._GetDevInfo(data)
self.failUnless(("local_dev" not in result and
"meta_dev" not in result and
def testParserDisk(self):
"""Test drbdsetup show parser for disk and network"""
- data = open("data/bdev-disk.txt").read()
+ data = self._get_contents("data/bdev-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")
+
+class TestDRBD8Status(unittest.TestCase):
+ """Testing case for DRBD8 /proc status"""
+
+ def setUp(self):
+ """Read in txt data"""
+ proc_data = "test/data/proc_drbd8.txt"
+ prefix = os.environ.get("srcdir", None)
+ if prefix:
+ proc_data = prefix + "/" + proc_data
+ self.proc_data = bdev.DRBD8._GetProcData(filename=proc_data)
+ self.mass_data = bdev.DRBD8._MassageProcData(self.proc_data)
+
+ def testMinorNotFound(self):
+ """Test not-found-minor in /proc"""
+ self.failUnless(9 not in self.mass_data)
+
+ def testLineNotMatch(self):
+ """Test wrong line passed to DRBD8Status"""
+ self.assertRaises(errors.BlockDeviceError, bdev.DRBD8Status, "foo")
+
+ def testMinor0(self):
+ """Test connected, primary device"""
+ stats = bdev.DRBD8Status(self.mass_data[0])
+ 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"""
+ stats = bdev.DRBD8Status(self.mass_data[1])
+ self.failUnless(stats.is_connected and stats.is_secondary and
+ stats.peer_primary and stats.is_disk_uptodate)
+
+ def testMinor4(self):
+ """Test WFconn device"""
+ stats = bdev.DRBD8Status(self.mass_data[4])
+ self.failUnless(stats.is_wfconn and stats.is_primary and
+ stats.rrole == 'Unknown' and
+ stats.is_disk_uptodate)
+
+ def testMinor6(self):
+ """Test diskless device"""
+ stats = bdev.DRBD8Status(self.mass_data[6])
+ self.failUnless(stats.is_connected and stats.is_secondary and
+ stats.peer_primary and stats.is_diskless)
+
+ def testMinor8(self):
+ """Test standalone device"""
+ stats = bdev.DRBD8Status(self.mass_data[8])
+ self.failUnless(stats.is_standalone and
+ stats.rrole == 'Unknown' and
+ stats.is_disk_uptodate)
+
if __name__ == '__main__':
unittest.main()