Convert snapshot_export rpc to new style

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

diff --git a/test/ganeti.bdev_unittest.py b/test/ganeti.bdev_unittest.py
index 7876f3d..b2299b1 100755 (executable)
--- 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()