Fix LUGrowDisk

[ganeti-local] / test / ganeti.bdev_unittest.py

diff --git a/test/ganeti.bdev_unittest.py b/test/ganeti.bdev_unittest.py
index 5a775c7..6973f17 100755 (executable)
--- a/test/ganeti.bdev_unittest.py
+++ b/test/ganeti.bdev_unittest.py
@@ -22,9 +22,11 @@
 """Script for unittesting the bdev module"""
 
 
+import os
 import unittest
 
 from ganeti import bdev
+from ganeti import errors
 
 
 class TestDRBD8Runner(unittest.TestCase):
@@ -44,6 +46,20 @@ 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 = (
@@ -60,7 +76,7 @@ class TestDRBD8Runner(unittest.TestCase):
 
   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"),
@@ -71,7 +87,7 @@ class TestDRBD8Runner(unittest.TestCase):
 
   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
@@ -83,7 +99,7 @@ class TestDRBD8Runner(unittest.TestCase):
 
   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"),
@@ -92,5 +108,58 @@ class TestDRBD8Runner(unittest.TestCase):
                      "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()