import warnings
import distutils.version
import glob
+import md5
+import errno
import ganeti
import testutils
ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress, \
TailFile, ForceDictType, SafeEncode, IsNormAbsPath, FormatTime, \
- UnescapeAndSplit, RunParts, PathJoin, HostInfo
+ UnescapeAndSplit, RunParts, PathJoin, HostInfo, ReadOneLineFile
from ganeti.errors import LockError, UnitParseError, GenericError, \
ProgrammerError, OpPrereqError
None)
+class TestReadFile(testutils.GanetiTestCase):
+
+ def testReadAll(self):
+ data = utils.ReadFile(self._TestDataFilename("cert1.pem"))
+ self.assertEqual(len(data), 814)
+
+ h = md5.new()
+ h.update(data)
+ self.assertEqual(h.hexdigest(), "a491efb3efe56a0535f924d5f8680fd4")
+
+ def testReadSize(self):
+ data = utils.ReadFile(self._TestDataFilename("cert1.pem"),
+ size=100)
+ self.assertEqual(len(data), 100)
+
+ h = md5.new()
+ h.update(data)
+ self.assertEqual(h.hexdigest(), "893772354e4e690b9efd073eed433ce7")
+
+ def testError(self):
+ self.assertRaises(EnvironmentError, utils.ReadFile,
+ "/dev/null/does-not-exist")
+
+
+class TestReadOneLineFile(testutils.GanetiTestCase):
+
+ def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
+
+ def testDefault(self):
+ data = ReadOneLineFile(self._TestDataFilename("cert1.pem"))
+ self.assertEqual(len(data), 27)
+ self.assertEqual(data, "-----BEGIN CERTIFICATE-----")
+
+ def testNotStrict(self):
+ data = ReadOneLineFile(self._TestDataFilename("cert1.pem"), strict=False)
+ self.assertEqual(len(data), 27)
+ self.assertEqual(data, "-----BEGIN CERTIFICATE-----")
+
+ def testStrictFailure(self):
+ self.assertRaises(errors.GenericError, ReadOneLineFile,
+ self._TestDataFilename("cert1.pem"), strict=True)
+
+ def testLongLine(self):
+ dummydata = (1024 * "Hello World! ")
+ myfile = self._CreateTempFile()
+ utils.WriteFile(myfile, data=dummydata)
+ datastrict = ReadOneLineFile(myfile, strict=True)
+ datalax = ReadOneLineFile(myfile, strict=False)
+ self.assertEqual(dummydata, datastrict)
+ self.assertEqual(dummydata, datalax)
+
+ def testNewline(self):
+ myfile = self._CreateTempFile()
+ myline = "myline"
+ for nl in ["", "\n", "\r\n"]:
+ dummydata = "%s%s" % (myline, nl)
+ utils.WriteFile(myfile, data=dummydata)
+ datalax = ReadOneLineFile(myfile, strict=False)
+ self.assertEqual(myline, datalax)
+ datastrict = ReadOneLineFile(myfile, strict=True)
+ self.assertEqual(myline, datastrict)
+
+ def testWhitespaceAndMultipleLines(self):
+ myfile = self._CreateTempFile()
+ for nl in ["", "\n", "\r\n"]:
+ for ws in [" ", "\t", "\t\t \t", "\t "]:
+ dummydata = (1024 * ("Foo bar baz %s%s" % (ws, nl)))
+ utils.WriteFile(myfile, data=dummydata)
+ datalax = ReadOneLineFile(myfile, strict=False)
+ if nl:
+ self.assert_(set("\r\n") & set(dummydata))
+ self.assertRaises(errors.GenericError, ReadOneLineFile,
+ myfile, strict=True)
+ explen = len("Foo bar baz ") + len(ws)
+ self.assertEqual(len(datalax), explen)
+ self.assertEqual(datalax, dummydata[:explen])
+ self.assertFalse(set("\r\n") & set(datalax))
+ else:
+ datastrict = ReadOneLineFile(myfile, strict=True)
+ self.assertEqual(dummydata, datastrict)
+ self.assertEqual(dummydata, datalax)
+
+ def testEmptylines(self):
+ myfile = self._CreateTempFile()
+ myline = "myline"
+ for nl in ["\n", "\r\n"]:
+ for ol in ["", "otherline"]:
+ dummydata = "%s%s%s%s%s%s" % (nl, nl, myline, nl, ol, nl)
+ utils.WriteFile(myfile, data=dummydata)
+ self.assert_(set("\r\n") & set(dummydata))
+ datalax = ReadOneLineFile(myfile, strict=False)
+ self.assertEqual(myline, datalax)
+ if ol:
+ self.assertRaises(errors.GenericError, ReadOneLineFile,
+ myfile, strict=True)
+ else:
+ datastrict = ReadOneLineFile(myfile, strict=True)
+ self.assertEqual(myline, datastrict)
+
+
class TestTimestampForFilename(unittest.TestCase):
def test(self):
self.assert_("." not in utils.TimestampForFilename())
def testNowOwnAddress(self):
"""check that I don't own an address"""
- # network 192.0.2.0/24 is reserved for test/documentation as per
- # rfc 3330, so we *should* not have an address of this range... if
+ # Network 192.0.2.0/24 is reserved for test/documentation as per
+ # RFC 5735, so we *should* not have an address of this range... if
# this fails, we should extend the test to multiple addresses
DST_IP = "192.0.2.1"
self.failIf(OwnIpAddress(DST_IP), "Should not own IP address %s" % DST_IP)
salt="xyz0"))
+class TestIgnoreSignals(unittest.TestCase):
+ """Test the IgnoreSignals decorator"""
+
+ @staticmethod
+ def _Raise(exception):
+ raise exception
+
+ @staticmethod
+ def _Return(rval):
+ return rval
+
+ def testIgnoreSignals(self):
+ sock_err_intr = socket.error(errno.EINTR, "Message")
+ sock_err_intr.errno = errno.EINTR
+ sock_err_inval = socket.error(errno.EINVAL, "Message")
+ sock_err_inval.errno = errno.EINVAL
+
+ env_err_intr = EnvironmentError(errno.EINTR, "Message")
+ env_err_inval = EnvironmentError(errno.EINVAL, "Message")
+
+ self.assertRaises(socket.error, self._Raise, sock_err_intr)
+ self.assertRaises(socket.error, self._Raise, sock_err_inval)
+ self.assertRaises(EnvironmentError, self._Raise, env_err_intr)
+ self.assertRaises(EnvironmentError, self._Raise, env_err_inval)
+
+ self.assertEquals(utils.IgnoreSignals(self._Raise, sock_err_intr), None)
+ self.assertEquals(utils.IgnoreSignals(self._Raise, env_err_intr), None)
+ self.assertRaises(socket.error, utils.IgnoreSignals, self._Raise,
+ sock_err_inval)
+ self.assertRaises(EnvironmentError, utils.IgnoreSignals, self._Raise,
+ env_err_inval)
+
+ self.assertEquals(utils.IgnoreSignals(self._Return, True), True)
+ self.assertEquals(utils.IgnoreSignals(self._Return, 33), 33)
+
+
if __name__ == '__main__':
testutils.GanetiTestProgram()
projects / ganeti-local / blobdiff