X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/6b90c22e1b8222796150f86467c8664677a344ad..cd098c41522469e462efeb1e157a8718ff7b7809:/test/ganeti.bdev_unittest.py diff --git a/test/ganeti.bdev_unittest.py b/test/ganeti.bdev_unittest.py index 6f0b098..b2299b1 100755 --- a/test/ganeti.bdev_unittest.py +++ b/test/ganeti.bdev_unittest.py @@ -25,11 +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 @@ -46,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 = ( @@ -75,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 @@ -100,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"), @@ -110,17 +107,30 @@ class TestDRBD8Runner(unittest.TestCase): "Should not find network info") -class TestDRBD8Status(unittest.TestCase): +class TestDRBD8Status(testutils.GanetiTestCase): """Testing case for DRBD8 /proc status""" def setUp(self): """Read in txt data""" - self.proc_data = bdev.DRBD8._GetProcData(filename="data/proc_drbd8.txt") + 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""" @@ -128,35 +138,51 @@ class TestDRBD8Status(unittest.TestCase): 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) + 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""" - 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) + 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""" - 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) + 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""" - stats = bdev.DRBD8Status(self.mass_data[6]) - self.failUnless(stats.is_connected and stats.is_secondary and - stats.peer_primary and stats.is_diskless) + 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""" - stats = bdev.DRBD8Status(self.mass_data[8]) - self.failUnless(stats.is_standalone and - stats.rrole == 'Unknown' and - stats.is_disk_uptodate) + 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()