Synnefo » snf-ganeti

  def testParser80(self):

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

    result = bdev.DRBD8._GetDevInfo(data)

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

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

  def testParser83(self):

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

    result = bdev.DRBD8._GetDevInfo(data)

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

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

  def testParserNetIP4(self):

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

    result = bdev.DRBD8._GetDevInfo(data)

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

                     "meta_dev" not in result and

  def testParserNetIP6(self):

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

    result = bdev.DRBD8._GetDevInfo(data)

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

                     "meta_dev" not in result and

  def testParserDisk(self):

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

    result = bdev.DRBD8._GetDevInfo(data)

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

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

  def setUp(self):

    """Read in txt data"""

    testutils.GanetiTestCase.setUp(self)

    proc83_sync_krnl_data = \

    self.proc_data = bdev.DRBD8._GetProcData(filename=proc_data)

    self.proc80e_data = bdev.DRBD8._GetProcData(filename=proc80e_data)

    self.proc83_data = bdev.DRBD8._GetProcData(filename=proc83_data)

  def testHelper(self):

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

    drbd_helper = bdev.DRBD8.GetUsermodeHelper(filename=sys_drbd_helper)

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

    shutil.rmtree(self.tmpdir)

  def testVerifyCertificate(self):

    nonexist_filename = os.path.join(self.tmpdir, "does-not-exist")

    self.assertEqual(errcode, cmdlib.LUClusterVerifyConfig.ETYPE_ERROR)

    # Try to load non-certificate file

    (errcode, msg) = cmdlib._VerifyCertificate(invalid_cert)

    self.assertEqual(errcode, cmdlib.LUClusterVerifyConfig.ETYPE_ERROR)

class TestVersionChecking(testutils.GanetiTestCase):

  def testParseVersion(self):

    parse = hv_kvm.KVMHypervisor._ParseKVMVersion

    self.assertEqual(parse(help_112), ("1.1.2", 1, 1, 2))

    self.assertEqual(parse(help_10), ("1.0", 1, 0, 0))

    self.assertEqual(parse(help_01590), ("0.15.90", 0, 15, 90))

    """

    boot_re = hv_kvm.KVMHypervisor._BOOT_RE

    self.assertTrue(boot_re.search(help_091))

    self.assertTrue(boot_re.search(help_0125))

    # IPv4-only, fake loopback interface

    tests = ["ip-addr-show-lo-ipv4.txt", "ip-addr-show-lo-oneline-ipv4.txt"]

    for test_file in tests:

      addr = netutils._GetIpAddressesFromIpOutput(data)

      self.failUnless(len(addr[4]) == 1 and addr[4][0] == "127.0.0.1" and not

                      addr[6])

    # IPv6-only, fake loopback interface

    tests = ["ip-addr-show-lo-ipv6.txt", "ip-addr-show-lo-ipv6.txt"]

    for test_file in tests:

      addr = netutils._GetIpAddressesFromIpOutput(data)

      self.failUnless(len(addr[6]) == 1 and addr[6][0] == "::1" and not addr[4])

    # IPv4 and IPv6, fake loopback interface

    tests = ["ip-addr-show-lo.txt", "ip-addr-show-lo-oneline.txt"]

    for test_file in tests:

      addr = netutils._GetIpAddressesFromIpOutput(data)

      self.failUnless(len(addr[6]) == 1 and addr[6][0] == "::1" and

                      len(addr[4]) == 1 and addr[4][0] == "127.0.0.1")

    # IPv4 and IPv6, dummy interface

    addr = netutils._GetIpAddressesFromIpOutput(data)

    self.failUnless(len(addr[6]) == 1 and

                    addr[6][0] == "2001:db8:85a3::8a2e:370:7334" and

  def testOldVersion(self):

    lvmvg = storage.LvmVgStorage()

    self.run_history = [

      ([self.VGREDUCE_CMD, "--removemissing", self.VGNAME],

       utils.RunResult(0, None, stdout, "", "", None, None)),

  def testNewVersion(self):

    lvmvg = storage.LvmVgStorage()

    # first: require --fail, check that it's used

    self.run_history = [

      ([self.VGREDUCE_CMD, "--removemissing", self.VGNAME],

      }, _verify_fn=lambda _: None)

  def testNoPrivateKey(self):

    cert_pem = utils.ReadFile(cert_filename)

    self.assertRaises(errors.X509CertError,

    assert cert.get_subject()

  def testSuccessfulCheck(self):

    cert_pem = utils.ReadFile(cert_filename)

    result = \

      node_daemon_setup._VerifyCertificate(cert_pem, _check_fn=self._Check)

    self.assertTrue(key)

  def testMismatchingKey(self):

    # Extract certificate

    cert1 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

    prepare_node_join.VerifyCertificate({}, _verify_fn=NotImplemented)

  def testGivenPrivateKey(self):

    cert_pem = utils.ReadFile(cert_filename)

    self.assertRaises(_JoinError, prepare_node_join._VerifyCertificate,

    assert cert.get_subject()

  def testSuccessfulCheck(self):

    cert_pem = utils.ReadFile(cert_filename)

    prepare_node_join._VerifyCertificate(cert_pem, _check_fn=self._Check)

class TestReadFile(testutils.GanetiTestCase):

  def testReadAll(self):

    self.assertEqual(len(data), 814)

    h = compat.md5_hash()

    self.assertEqual(h.hexdigest(), "a491efb3efe56a0535f924d5f8680fd4")

  def testReadSize(self):

                          size=100)

    self.assertEqual(len(data), 100)

  def testCallback(self):

    def _Cb(fh):

      self.assertEqual(fh.tell(), 0)

    self.assertEqual(len(data), 814)

  def testError(self):

    testutils.GanetiTestCase.setUp(self)

  def testDefault(self):

    self.assertEqual(len(data), 27)

    self.assertEqual(data, "-----BEGIN CERTIFICATE-----")

  def testNotStrict(self):

                                 strict=False)

    self.assertEqual(len(data), 27)

    self.assertEqual(data, "-----BEGIN CERTIFICATE-----")

  def testStrictFailure(self):

    self.assertRaises(errors.GenericError, utils.ReadOneLineFile,

  def testLongLine(self):

    dummydata = (1024 * "Hello World! ")

      temp.close()

  def testNoInputRead(self):

    result = utils.RunCmd(["cat"], timeout=10.0)

    self.assertFalse(result.failed)

    self.assertEqual(result.stdout, "")

  def testInputFileHandle(self):

    result = utils.RunCmd(["cat"], input_fd=open(testfile, "r"))

    self.assertFalse(result.failed)

    self.assertEqual(result.stderr, "")

  def testInputNumericFileDescriptor(self):

    fh = open(testfile, "r")

    try:

    self.assertEqual(result.stderr, "")

  def testInputWithCloseFds(self):

    temp = open(self.fname, "r+")

    try:

  def _LoadCert(self, name):

    return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

  def test(self):

    validity = utils.GetX509CertValidity(self._LoadCert("cert1.pem"))

    shutil.rmtree(self.tmpdir)

  def testVerifyCertificate(self):

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                           cert_pem)

    shutil.rmtree(self.tmpdir)

  def testMismatchingKey(self):

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

                                           utils.ReadFile(other_cert))

      self.fail("Exception was not raised")

  def testMatchingKey(self):

    # Extract certificate

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,

    utils.CheckNodeCertificate(cert, _noded_cert_file=cert_filename)

  def testMissingFile(self):

    nodecert = utils.PathJoin(self.tmpdir, "does-not-exist")

    utils.CheckNodeCertificate(NotImplemented, _noded_cert_file=nodecert)

                      NotImplemented, _noded_cert_file=tmpfile)

  def testNoPrivateKey(self):

    self.assertRaises(errors.X509CertError, utils.CheckNodeCertificate,

                      NotImplemented, _noded_cert_file=cert)

  def testMismatchInNodeCert(self):

    tmpfile = utils.PathJoin(self.tmpdir, "cert")

    # Extract certificate

  return os.environ.get("TOP_SRCDIR", ".")

def _SetupLogging(verbose):

  """Setupup logging infrastructure.

                            UnifyValueType(second),

                            msg=msg)

  def _CreateTempFile(self):

    """Creates a temporary file and adds it to the internal cleanup list.