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 |