1 #!/usr/bin/env python2 2 3 # Copyright 2016 The Chromium OS Authors. All rights reserved. 4 # Use of this source code is governed by a BSD-style license that can be 5 # found in the LICENSE file. 6 7 """Unit tests for apache_log_metrics.py""" 8 9 from __future__ import print_function 10 11 import StringIO 12 import subprocess 13 import tempfile 14 import threading 15 import time 16 import unittest 17 18 import tail_until_writer_finished 19 20 21 class TestTailUntilWriterFinished(unittest.TestCase): 22 """Tests tail_until_writer_finished.""" 23 24 def SkipIfMissingInotifyTools(self): 25 """The tail_until_writer_finished module requires 'inotifywait'.""" 26 try: 27 subprocess.call(['inotifywait'], stderr=subprocess.PIPE) 28 except OSError: 29 raise unittest.SkipTest('inotify-tools must be installed.') 30 31 def testTail(self): 32 """Tests reading a file from the end.""" 33 self.GetsEntireInput(seek_to_end=True) 34 35 def testRead(self): 36 """Tests reading a file from the beginning.""" 37 self.GetsEntireInput(seek_to_end=False) 38 39 def GetsEntireInput(self, seek_to_end): 40 """Tails a temp file in a thread. 41 42 Check that it read the file correctly. 43 44 @param seek_to_end: Whether to .seek to the end of the file before 45 reading. 46 """ 47 self.SkipIfMissingInotifyTools() 48 49 f = tempfile.NamedTemporaryFile() 50 output = StringIO.StringIO() 51 52 f.write('This line will not get read if we seek to end.\n') 53 f.flush() 54 55 def Tail(): 56 """Tails the file into |output| with a 64k chunk size.""" 57 tail_until_writer_finished.TailFile( 58 f.name, 0.1, 64000, outfile=output, seek_to_end=seek_to_end) 59 60 thread = threading.Thread(target=Tail) 61 thread.start() 62 63 # There is a race here: the thread must start the inotify process before 64 # we close the file. This shouldn't take long at all, so add a small 65 # sleep. 66 time.sleep(0.3) 67 68 for i in range(100): 69 f.write(str(i) + '\n') 70 f.flush() 71 f.close() 72 thread.join() 73 74 expected = ''.join([str(i) + '\n' for i in range(100)]) 75 if not seek_to_end: 76 expected = ('This line will not get read if we seek to end.\n' 77 + expected) 78 self.assertEqual(output.getvalue(), expected) 79 80 81 if __name__ == '__main__': 82 unittest.main() 83