| OLD | NEW |
| (Empty) |
| 1 # Copyright (c) 2001-2007 Twisted Matrix Laboratories. | |
| 2 # See LICENSE for details. | |
| 3 | |
| 4 from twisted.trial import unittest | |
| 5 | |
| 6 # system imports | |
| 7 import os, time, stat | |
| 8 | |
| 9 # twisted imports | |
| 10 from twisted.python import logfile, runtime | |
| 11 | |
| 12 | |
| 13 class LogFileTestCase(unittest.TestCase): | |
| 14 """ | |
| 15 Test the rotating log file. | |
| 16 """ | |
| 17 | |
| 18 def setUp(self): | |
| 19 self.dir = self.mktemp() | |
| 20 os.makedirs(self.dir) | |
| 21 self.name = "test.log" | |
| 22 self.path = os.path.join(self.dir, self.name) | |
| 23 | |
| 24 | |
| 25 def tearDown(self): | |
| 26 """ | |
| 27 Restore back write rights on created paths: if tests modified the | |
| 28 rights, that will allow the paths to be removed easily afterwards. | |
| 29 """ | |
| 30 os.chmod(self.dir, 0777) | |
| 31 if os.path.exists(self.path): | |
| 32 os.chmod(self.path, 0777) | |
| 33 | |
| 34 | |
| 35 def testWriting(self): | |
| 36 log = logfile.LogFile(self.name, self.dir) | |
| 37 log.write("123") | |
| 38 log.write("456") | |
| 39 log.flush() | |
| 40 log.write("7890") | |
| 41 log.close() | |
| 42 | |
| 43 f = open(self.path, "r") | |
| 44 self.assertEquals(f.read(), "1234567890") | |
| 45 f.close() | |
| 46 | |
| 47 def testRotation(self): | |
| 48 # this logfile should rotate every 10 bytes | |
| 49 log = logfile.LogFile(self.name, self.dir, rotateLength=10) | |
| 50 | |
| 51 # test automatic rotation | |
| 52 log.write("123") | |
| 53 log.write("4567890") | |
| 54 log.write("1" * 11) | |
| 55 self.assert_(os.path.exists("%s.1" % self.path)) | |
| 56 self.assert_(not os.path.exists("%s.2" % self.path)) | |
| 57 log.write('') | |
| 58 self.assert_(os.path.exists("%s.1" % self.path)) | |
| 59 self.assert_(os.path.exists("%s.2" % self.path)) | |
| 60 self.assert_(not os.path.exists("%s.3" % self.path)) | |
| 61 log.write("3") | |
| 62 self.assert_(not os.path.exists("%s.3" % self.path)) | |
| 63 | |
| 64 # test manual rotation | |
| 65 log.rotate() | |
| 66 self.assert_(os.path.exists("%s.3" % self.path)) | |
| 67 self.assert_(not os.path.exists("%s.4" % self.path)) | |
| 68 log.close() | |
| 69 | |
| 70 self.assertEquals(log.listLogs(), [1, 2, 3]) | |
| 71 | |
| 72 def testAppend(self): | |
| 73 log = logfile.LogFile(self.name, self.dir) | |
| 74 log.write("0123456789") | |
| 75 log.close() | |
| 76 | |
| 77 log = logfile.LogFile(self.name, self.dir) | |
| 78 self.assertEquals(log.size, 10) | |
| 79 self.assertEquals(log._file.tell(), log.size) | |
| 80 log.write("abc") | |
| 81 self.assertEquals(log.size, 13) | |
| 82 self.assertEquals(log._file.tell(), log.size) | |
| 83 f = log._file | |
| 84 f.seek(0, 0) | |
| 85 self.assertEquals(f.read(), "0123456789abc") | |
| 86 log.close() | |
| 87 | |
| 88 def testLogReader(self): | |
| 89 log = logfile.LogFile(self.name, self.dir) | |
| 90 log.write("abc\n") | |
| 91 log.write("def\n") | |
| 92 log.rotate() | |
| 93 log.write("ghi\n") | |
| 94 log.flush() | |
| 95 | |
| 96 # check reading logs | |
| 97 self.assertEquals(log.listLogs(), [1]) | |
| 98 reader = log.getCurrentLog() | |
| 99 reader._file.seek(0) | |
| 100 self.assertEquals(reader.readLines(), ["ghi\n"]) | |
| 101 self.assertEquals(reader.readLines(), []) | |
| 102 reader.close() | |
| 103 reader = log.getLog(1) | |
| 104 self.assertEquals(reader.readLines(), ["abc\n", "def\n"]) | |
| 105 self.assertEquals(reader.readLines(), []) | |
| 106 reader.close() | |
| 107 | |
| 108 # check getting illegal log readers | |
| 109 self.assertRaises(ValueError, log.getLog, 2) | |
| 110 self.assertRaises(TypeError, log.getLog, "1") | |
| 111 | |
| 112 # check that log numbers are higher for older logs | |
| 113 log.rotate() | |
| 114 self.assertEquals(log.listLogs(), [1, 2]) | |
| 115 reader = log.getLog(1) | |
| 116 reader._file.seek(0) | |
| 117 self.assertEquals(reader.readLines(), ["ghi\n"]) | |
| 118 self.assertEquals(reader.readLines(), []) | |
| 119 reader.close() | |
| 120 reader = log.getLog(2) | |
| 121 self.assertEquals(reader.readLines(), ["abc\n", "def\n"]) | |
| 122 self.assertEquals(reader.readLines(), []) | |
| 123 reader.close() | |
| 124 | |
| 125 def testModePreservation(self): | |
| 126 """ | |
| 127 Check rotated files have same permissions as original. | |
| 128 """ | |
| 129 f = open(self.path, "w").close() | |
| 130 os.chmod(self.path, 0707) | |
| 131 mode = os.stat(self.path)[stat.ST_MODE] | |
| 132 log = logfile.LogFile(self.name, self.dir) | |
| 133 log.write("abc") | |
| 134 log.rotate() | |
| 135 self.assertEquals(mode, os.stat(self.path)[stat.ST_MODE]) | |
| 136 | |
| 137 | |
| 138 def test_noPermission(self): | |
| 139 """ | |
| 140 Check it keeps working when permission on dir changes. | |
| 141 """ | |
| 142 log = logfile.LogFile(self.name, self.dir) | |
| 143 log.write("abc") | |
| 144 | |
| 145 # change permissions so rotation would fail | |
| 146 os.chmod(self.dir, 0444) | |
| 147 | |
| 148 # if this succeeds, chmod doesn't restrict us, so we can't | |
| 149 # do the test | |
| 150 try: | |
| 151 f = open(os.path.join(self.dir,"xxx"), "w") | |
| 152 except (OSError, IOError): | |
| 153 pass | |
| 154 else: | |
| 155 f.close() | |
| 156 return | |
| 157 | |
| 158 log.rotate() # this should not fail | |
| 159 | |
| 160 log.write("def") | |
| 161 log.flush() | |
| 162 | |
| 163 f = log._file | |
| 164 self.assertEquals(f.tell(), 6) | |
| 165 f.seek(0, 0) | |
| 166 self.assertEquals(f.read(), "abcdef") | |
| 167 log.close() | |
| 168 | |
| 169 | |
| 170 def test_maxNumberOfLog(self): | |
| 171 """ | |
| 172 Test it respect the limit on the number of files when maxRotatedFiles | |
| 173 is not None. | |
| 174 """ | |
| 175 log = logfile.LogFile(self.name, self.dir, rotateLength=10, | |
| 176 maxRotatedFiles=3) | |
| 177 log.write("1" * 11) | |
| 178 log.write("2" * 11) | |
| 179 self.failUnless(os.path.exists("%s.1" % self.path)) | |
| 180 | |
| 181 log.write("3" * 11) | |
| 182 self.failUnless(os.path.exists("%s.2" % self.path)) | |
| 183 | |
| 184 log.write("4" * 11) | |
| 185 self.failUnless(os.path.exists("%s.3" % self.path)) | |
| 186 self.assertEquals(file("%s.3" % self.path).read(), "1" * 11) | |
| 187 | |
| 188 log.write("5" * 11) | |
| 189 self.assertEquals(file("%s.3" % self.path).read(), "2" * 11) | |
| 190 self.failUnless(not os.path.exists("%s.4" % self.path)) | |
| 191 | |
| 192 def test_fromFullPath(self): | |
| 193 """ | |
| 194 Test the fromFullPath method. | |
| 195 """ | |
| 196 log1 = logfile.LogFile(self.name, self.dir, 10, defaultMode=0777) | |
| 197 log2 = logfile.LogFile.fromFullPath(self.path, 10, defaultMode=0777) | |
| 198 self.assertEquals(log1.name, log2.name) | |
| 199 self.assertEquals(os.path.abspath(log1.path), log2.path) | |
| 200 self.assertEquals(log1.rotateLength, log2.rotateLength) | |
| 201 self.assertEquals(log1.defaultMode, log2.defaultMode) | |
| 202 | |
| 203 def test_defaultPermissions(self): | |
| 204 """ | |
| 205 Test the default permission of the log file: if the file exist, it | |
| 206 should keep the permission. | |
| 207 """ | |
| 208 f = file(self.path, "w") | |
| 209 os.chmod(self.path, 0707) | |
| 210 currentMode = stat.S_IMODE(os.stat(self.path)[stat.ST_MODE]) | |
| 211 f.close() | |
| 212 log1 = logfile.LogFile(self.name, self.dir) | |
| 213 self.assertEquals(stat.S_IMODE(os.stat(self.path)[stat.ST_MODE]), | |
| 214 currentMode) | |
| 215 | |
| 216 def test_specifiedPermissions(self): | |
| 217 """ | |
| 218 Test specifying the permissions used on the log file. | |
| 219 """ | |
| 220 log1 = logfile.LogFile(self.name, self.dir, defaultMode=0066) | |
| 221 mode = stat.S_IMODE(os.stat(self.path)[stat.ST_MODE]) | |
| 222 if runtime.platform.isWindows(): | |
| 223 # The only thing we can get here is global read-only | |
| 224 self.assertEquals(mode, 0444) | |
| 225 else: | |
| 226 self.assertEquals(mode, 0066) | |
| 227 | |
| 228 | |
| 229 class RiggedDailyLogFile(logfile.DailyLogFile): | |
| 230 _clock = 0.0 | |
| 231 | |
| 232 def _openFile(self): | |
| 233 logfile.DailyLogFile._openFile(self) | |
| 234 # rig the date to match _clock, not mtime | |
| 235 self.lastDate = self.toDate() | |
| 236 | |
| 237 def toDate(self, *args): | |
| 238 if args: | |
| 239 return time.gmtime(*args)[:3] | |
| 240 return time.gmtime(self._clock)[:3] | |
| 241 | |
| 242 class DailyLogFileTestCase(unittest.TestCase): | |
| 243 """ | |
| 244 Test rotating log file. | |
| 245 """ | |
| 246 | |
| 247 def setUp(self): | |
| 248 self.dir = self.mktemp() | |
| 249 os.makedirs(self.dir) | |
| 250 self.name = "testdaily.log" | |
| 251 self.path = os.path.join(self.dir, self.name) | |
| 252 | |
| 253 | |
| 254 def testWriting(self): | |
| 255 log = RiggedDailyLogFile(self.name, self.dir) | |
| 256 log.write("123") | |
| 257 log.write("456") | |
| 258 log.flush() | |
| 259 log.write("7890") | |
| 260 log.close() | |
| 261 | |
| 262 f = open(self.path, "r") | |
| 263 self.assertEquals(f.read(), "1234567890") | |
| 264 f.close() | |
| 265 | |
| 266 def testRotation(self): | |
| 267 # this logfile should rotate every 10 bytes | |
| 268 log = RiggedDailyLogFile(self.name, self.dir) | |
| 269 days = [(self.path + '.' + log.suffix(day * 86400)) for day in range(3)] | |
| 270 | |
| 271 # test automatic rotation | |
| 272 log._clock = 0.0 # 1970/01/01 00:00.00 | |
| 273 log.write("123") | |
| 274 log._clock = 43200 # 1970/01/01 12:00.00 | |
| 275 log.write("4567890") | |
| 276 log._clock = 86400 # 1970/01/02 00:00.00 | |
| 277 log.write("1" * 11) | |
| 278 self.assert_(os.path.exists(days[0])) | |
| 279 self.assert_(not os.path.exists(days[1])) | |
| 280 log._clock = 172800 # 1970/01/03 00:00.00 | |
| 281 log.write('') | |
| 282 self.assert_(os.path.exists(days[0])) | |
| 283 self.assert_(os.path.exists(days[1])) | |
| 284 self.assert_(not os.path.exists(days[2])) | |
| 285 log._clock = 259199 # 1970/01/03 23:59.59 | |
| 286 log.write("3") | |
| 287 self.assert_(not os.path.exists(days[2])) | |
| 288 | |
| OLD | NEW |