← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_logfile.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_logfile.py
 
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, 0555)
 
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
 
 
217
    def test_specifiedPermissions(self):
 
218
        """
 
219
        Test specifying the permissions used on the log file.
 
220
        """
 
221
        log1 = logfile.LogFile(self.name, self.dir, defaultMode=0066)
 
222
        mode = stat.S_IMODE(os.stat(self.path)[stat.ST_MODE])
 
223
        if runtime.platform.isWindows():
 
224
            # The only thing we can get here is global read-only
 
225
            self.assertEquals(mode, 0444)
 
226
        else:
 
227
            self.assertEquals(mode, 0066)
 
228
 
 
229
 
 
230
    def test_reopen(self):
 
231
        """
 
232
        L{logfile.LogFile.reopen} allows to rename the currently used file and
 
233
        make L{logfile.LogFile} create a new file.
 
234
        """
 
235
        log1 = logfile.LogFile(self.name, self.dir)
 
236
        log1.write("hello1")
 
237
        savePath = os.path.join(self.dir, "save.log")
 
238
        os.rename(self.path, savePath)
 
239
        log1.reopen()
 
240
        log1.write("hello2")
 
241
        log1.close()
 
242
 
 
243
        f = open(self.path, "r")
 
244
        self.assertEquals(f.read(), "hello2")
 
245
        f.close()
 
246
        f = open(savePath, "r")
 
247
        self.assertEquals(f.read(), "hello1")
 
248
        f.close()
 
249
 
 
250
    if runtime.platform.isWindows():
 
251
        test_reopen.skip = "Can't test reopen on Windows"
 
252
 
 
253
 
 
254
 
 
255
class RiggedDailyLogFile(logfile.DailyLogFile):
 
256
    _clock = 0.0
 
257
 
 
258
    def _openFile(self):
 
259
        logfile.DailyLogFile._openFile(self)
 
260
        # rig the date to match _clock, not mtime
 
261
        self.lastDate = self.toDate()
 
262
 
 
263
    def toDate(self, *args):
 
264
        if args:
 
265
            return time.gmtime(*args)[:3]
 
266
        return time.gmtime(self._clock)[:3]
 
267
 
 
268
class DailyLogFileTestCase(unittest.TestCase):
 
269
    """
 
270
    Test rotating log file.
 
271
    """
 
272
 
 
273
    def setUp(self):
 
274
        self.dir = self.mktemp()
 
275
        os.makedirs(self.dir)
 
276
        self.name = "testdaily.log"
 
277
        self.path = os.path.join(self.dir, self.name)
 
278
 
 
279
 
 
280
    def testWriting(self):
 
281
        log = RiggedDailyLogFile(self.name, self.dir)
 
282
        log.write("123")
 
283
        log.write("456")
 
284
        log.flush()
 
285
        log.write("7890")
 
286
        log.close()
 
287
 
 
288
        f = open(self.path, "r")
 
289
        self.assertEquals(f.read(), "1234567890")
 
290
        f.close()
 
291
 
 
292
    def testRotation(self):
 
293
        # this logfile should rotate every 10 bytes
 
294
        log = RiggedDailyLogFile(self.name, self.dir)
 
295
        days = [(self.path + '.' + log.suffix(day * 86400)) for day in range(3)]
 
296
 
 
297
        # test automatic rotation
 
298
        log._clock = 0.0    # 1970/01/01 00:00.00
 
299
        log.write("123")
 
300
        log._clock = 43200  # 1970/01/01 12:00.00
 
301
        log.write("4567890")
 
302
        log._clock = 86400  # 1970/01/02 00:00.00
 
303
        log.write("1" * 11)
 
304
        self.assert_(os.path.exists(days[0]))
 
305
        self.assert_(not os.path.exists(days[1]))
 
306
        log._clock = 172800 # 1970/01/03 00:00.00
 
307
        log.write('')
 
308
        self.assert_(os.path.exists(days[0]))
 
309
        self.assert_(os.path.exists(days[1]))
 
310
        self.assert_(not os.path.exists(days[2]))
 
311
        log._clock = 259199 # 1970/01/03 23:59.59
 
312
        log.write("3")
 
313
        self.assert_(not os.path.exists(days[2]))
 
314