Home | History | Annotate | Download | only in stats 
      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