twisted/test/test_logfile.py

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