X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/6634816b0954389cd54e1010f2eef1db0401ee40..ba55d062da8dfb89a37afc2f13f2e689d0094829:/test/ganeti.bdev_unittest.py diff --git a/test/ganeti.bdev_unittest.py b/test/ganeti.bdev_unittest.py index 7876f3d..b2299b1 100755 --- a/test/ganeti.bdev_unittest.py +++ b/test/ganeti.bdev_unittest.py @@ -25,10 +25,12 @@ import os import unittest +import testutils from ganeti import bdev +from ganeti import errors -class TestDRBD8Runner(unittest.TestCase): +class TestDRBD8Runner(testutils.GanetiTestCase): """Testing case for DRBD8""" @staticmethod @@ -45,21 +47,6 @@ 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 + "/" + 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 = ( @@ -74,20 +61,31 @@ class TestDRBD8Runner(unittest.TestCase): """Test drbdsetup show parser creation""" bdev.DRBD8._GetShowParser() - def testParserBoth(self): + def testParserBoth80(self): """Test drbdsetup show parser for disk and network""" - data = self._get_contents("data/bdev-both.txt") + data = self._ReadTestData("bdev-both.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") + "Wrong network info (8.0.x)") + + def testParserBoth83(self): + """Test drbdsetup show parser for disk and network""" + data = self._ReadTestData("bdev-8.3-both.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 (8.2.x)") def testParserNet(self): """Test drbdsetup show parser for disk and network""" - data = self._get_contents("data/bdev-net.txt") + data = self._ReadTestData("bdev-net.txt") result = bdev.DRBD8._GetDevInfo(data) self.failUnless(("local_dev" not in result and "meta_dev" not in result and @@ -99,7 +97,7 @@ class TestDRBD8Runner(unittest.TestCase): def testParserDisk(self): """Test drbdsetup show parser for disk and network""" - data = self._get_contents("data/bdev-disk.txt") + data = self._ReadTestData("bdev-disk.txt") result = bdev.DRBD8._GetDevInfo(data) self.failUnless(self._has_disk(result, "/dev/xenvg/test.data", "/dev/xenvg/test.meta"), @@ -108,5 +106,83 @@ class TestDRBD8Runner(unittest.TestCase): "remote_addr" not in result), "Should not find network info") + +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") + proc83_data = self._TestDataFilename("proc_drbd83.txt") + self.proc_data = bdev.DRBD8._GetProcData(filename=proc_data) + self.proc83_data = bdev.DRBD8._GetProcData(filename=proc83_data) + self.mass_data = bdev.DRBD8._MassageProcData(self.proc_data) + self.mass83_data = bdev.DRBD8._MassageProcData(self.proc83_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 testMinorNotFound(self): + """Test not-found-minor in /proc""" + self.failUnless(9 not in self.mass_data) + self.failUnless(9 not in self.mass83_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]: + 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) + if __name__ == '__main__': unittest.main()