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
class TestReadFile(testutils.GanetiTestCase):
- def setUp(self):
- testutils.GanetiTestCase.setUp(self)
-
- self.tmpdir = tempfile.mkdtemp()
- self.fname = utils.PathJoin(self.tmpdir, "data1")
-
- def tearDown(self):
- testutils.GanetiTestCase.tearDown(self)
-
- shutil.rmtree(self.tmpdir)
def testReadAll(self):
data = utils.ReadFile(self._TestDataFilename("cert1.pem"))
h.update(data)
self.assertEqual(h.hexdigest(), "893772354e4e690b9efd073eed433ce7")
- def testReadOneline(self):
- data = utils.ReadFile(self._TestDataFilename("cert1.pem"),
- oneline=True)
- self.assertEqual(len(data), 27)
- self.assertEqual(data, "-----BEGIN CERTIFICATE-----")
-
- def testReadOnelineSize(self):
- dummydata = (1024 * "Hello World! ")
- self.assertFalse(set("\r\n") & set(dummydata))
-
- utils.WriteFile(self.fname, data=dummydata)
-
- data = utils.ReadFile(self.fname, oneline=True, size=555)
- self.assertEqual(len(data), 555)
- self.assertEqual(data, dummydata[:555])
- self.assertFalse(set("\r\n") & set(data))
+ def testError(self):
+ self.assertRaises(EnvironmentError, utils.ReadFile,
+ "/dev/null/does-not-exist")
- def testReadOnelineSize2(self):
- for end in ["\n", "\r\n"]:
- dummydata = (1024 * ("Hello World%s" % end))
- self.assert_(set("\r\n") & set(dummydata))
- utils.WriteFile(self.fname, data=dummydata)
+class TestReadOneLineFile(testutils.GanetiTestCase):
- data = utils.ReadFile(self.fname, oneline=True, size=555)
- self.assertEqual(len(data), len("Hello World"))
- self.assertEqual(data, dummydata[:11])
- self.assertFalse(set("\r\n") & set(data))
+ def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
- def testReadOnelineWhitespace(self):
- for ws in [" ", "\t", "\t\t \t", "\t "]:
- dummydata = (1024 * ("Foo bar baz %s\n" % ws))
- self.assert_(set("\r\n") & set(dummydata))
+ def testDefault(self):
+ data = ReadOneLineFile(self._TestDataFilename("cert1.pem"))
+ self.assertEqual(len(data), 27)
+ self.assertEqual(data, "-----BEGIN CERTIFICATE-----")
- utils.WriteFile(self.fname, data=dummydata)
+ def testNotStrict(self):
+ data = ReadOneLineFile(self._TestDataFilename("cert1.pem"), strict=False)
+ self.assertEqual(len(data), 27)
+ self.assertEqual(data, "-----BEGIN CERTIFICATE-----")
- data = utils.ReadFile(self.fname, oneline=True, size=555)
- explen = len("Foo bar baz ") + len(ws)
- self.assertEqual(len(data), explen)
- self.assertEqual(data, dummydata[:explen])
- self.assertFalse(set("\r\n") & set(data))
+ def testStrictFailure(self):
+ self.assertRaises(errors.GenericError, ReadOneLineFile,
+ self._TestDataFilename("cert1.pem"), strict=True)
- def testError(self):
- self.assertRaises(EnvironmentError, utils.ReadFile,
- utils.PathJoin(self.tmpdir, "does-not-exist"))
+ 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 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