Synnefo » snf-ganeti

from ganeti import objects

from ganeti import utils

from ganeti.block import bdev

import testutils

      }

    ]

    for d,r in zip(data, result):

class TestDRBD8Runner(testutils.GanetiTestCase):

  @staticmethod

  def _has_disk(data, dname, mname):

  def testParserCreation(self):

    """Test drbdsetup show parser creation"""

  def testParser80(self):

    """Test drbdsetup show parser for disk and network version 8.0"""

    data = testutils.ReadTestData("bdev-drbd-8.0.txt")

    self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",

                                   "/dev/xenvg/test.meta"),

                    "Wrong local disk info")

  def testParser83(self):

    """Test drbdsetup show parser for disk and network version 8.3"""

    data = testutils.ReadTestData("bdev-drbd-8.3.txt")

    self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",

                                   "/dev/xenvg/test.meta"),

                    "Wrong local disk info")

  def testParserNetIP4(self):

    """Test drbdsetup show parser for IPv4 network"""

    data = testutils.ReadTestData("bdev-drbd-net-ip4.txt")

    self.failUnless(("local_dev" not in result and

                     "meta_dev" not in result and

                     "meta_index" not in result),

  def testParserNetIP6(self):

    """Test drbdsetup show parser for IPv6 network"""

    data = testutils.ReadTestData("bdev-drbd-net-ip6.txt")

    self.failUnless(("local_dev" not in result and

                     "meta_dev" not in result and

                     "meta_index" not in result),

  def testParserDisk(self):

    """Test drbdsetup show parser for disk"""

    data = testutils.ReadTestData("bdev-drbd-disk.txt")

    self.failUnless(self._has_disk(result, "/dev/xenvg/test.data",

                                   "/dev/xenvg/test.meta"),

                    "Wrong local disk info")

    for vmaj, vmin, vrel, opts, meta in should_fail:

      self.assertRaises(errors.BlockDeviceError,

                        vmaj, vmin, vrel, opts, meta)

    # get the valid options from the frozenset(frozenset()) in constants.

    for vmaj, vmin, vrel in ((8, 0, 0), (8, 0, 11), (8, 2, 6)):

      for opts in valid_options:

        self.assertRaises(errors.BlockDeviceError,

                          vmaj, vmin, vrel, opts, True)

    # Versions with partial support (testing only options that are supported)

    for test in tests:

      vmaj, vmin, vrel, disabled_barriers, disable_meta_flush, expected = test

      args = \

                                           disabled_barriers,

                                           disable_meta_flush)

      self.failUnless(set(args) == set(expected),

    # Unsupported or invalid versions

    for vmaj, vmin, vrel in ((0, 7, 25), (9, 0, 0), (7, 0, 0), (8, 4, 0)):

      self.assertRaises(errors.BlockDeviceError,

                        vmaj, vmin, vrel, "n", True)

    # Invalid options

    for option in ("", "c", "whatever", "nbdfc", "nf"):

      self.assertRaises(errors.BlockDeviceError,

                        8, 3, 11, option, True)

    proc83_sync_data = testutils.TestDataFilename("proc_drbd83_sync.txt")

    proc83_sync_krnl_data = \

      testutils.TestDataFilename("proc_drbd83_sync_krnl2.6.39.txt")

    self.proc83_sync_krnl_data = \

    self.mass83_sync_krnl_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,

  def testHelper(self):

    """Test reading usermode_helper in /sys."""

    sys_drbd_helper = testutils.TestDataFilename("sys_drbd_usermode_helper.txt")

    self.failUnlessEqual(drbd_helper, "/bin/true")

  def testHelperIOErrors(self):

    temp_file = self._CreateTempFile()

    os.unlink(temp_file)

    self.failUnlessRaises(errors.BlockDeviceError,

  def testMinorNotFound(self):

    """Test not-found-minor in /proc"""

    self.failUnless(3 not in self.mass80e_data)

  def testLineNotMatch(self):

  def testMinor0(self):

    """Test connected, primary device"""

    for data in [self.mass_data, self.mass83_data]:

      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]:

      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, self.mass80e_data]:

      self.failIf(stats.is_in_use)

  def testMinor4(self):

    """Test WFconn device"""

    for data in [self.mass_data, self.mass83_data]:

      self.failUnless(stats.is_in_use)

      self.failUnless(stats.is_wfconn and stats.is_primary and

                      stats.rrole == "Unknown" and

  def testMinor6(self):

    """Test diskless device"""

    for data in [self.mass_data, self.mass83_data]:

      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]:

      self.failUnless(stats.is_in_use)

      self.failUnless(stats.is_standalone and

                      stats.rrole == "Unknown" and

                      stats.is_disk_uptodate)

  def testDRBD83SyncFine(self):

    self.failUnless(stats.is_in_resync)

    self.failUnless(stats.sync_percent is not None)

  def testDRBD83SyncBroken(self):

    self.failUnless(stats.is_in_resync)

    self.failUnless(stats.sync_percent is not None)