ppf/new: Complete tests for util.py.

diff --git a/ppf/new/tests/test_util.py b/ppf/new/tests/test_util.py
index 8fb7289..baeaa67 100644 (file)
--- a/ppf/new/tests/test_util.py
+++ b/ppf/new/tests/test_util.py
@@ -27,12 +27,30 @@ class LiveLogFileTest(unittest.TestCase):
     ok_log_filename = "ok.log"
     ok_status_filename = "ok.log.status"
     empty_log_filename = "empty.log"
+    empty_status_filename = "empty.log.status"
     incomplete_log_filename = "incomplete.log"
+    incomplete_status_filename = "incomplete.log.status"
     no_log_filename = "no.log"
     first_line = "line 1\n"
     second_line = "line 1\n"
     third_line = "line 1\n"
 
+    def remove_status_files(self):
+        try:
+            os.remove(self.ok_status_filename)
+        except OSError, e:
+            pass
+
+        try:
+            os.remove(self.empty_status_filename)
+        except OSError, e:
+            pass
+
+        try:
+            os.remove(self.incomplete_status_filename)
+        except OSError, e:
+            pass
+
     def setUp(self):
         """Create test log files."""
         f = open(self.ok_log_filename, "wt")
@@ -48,16 +66,12 @@ class LiveLogFileTest(unittest.TestCase):
         f.write("no newline")
         f.close()
 
-        # Remove "no log" file and status file in case they exist.
+        # Remove "no log" file and status files in case they exist.
         try:
             os.remove(self.no_log_filename)
         except OSError, e:
             pass
-
-        try:
-            os.remove(self.ok_status_filename)
-        except OSError, e:
-            pass
+        self.remove_status_files()
 
     def tearDown(self):
         """Remove test log files."""
@@ -65,6 +79,13 @@ class LiveLogFileTest(unittest.TestCase):
         os.remove(self.empty_log_filename)
         os.remove(self.incomplete_log_filename)
 
+        # Remove "no log" file and status files in case they exist.
+        try:
+            os.remove(self.no_log_filename)
+        except OSError, e:
+            pass
+        self.remove_status_files()
+
     def read_position_in_status_file(self, status_filename):
         """Read first line in status file as integer."""
         f = open(status_filename, "rt")
@@ -90,8 +111,8 @@ class LiveLogFileTest(unittest.TestCase):
 
         # Log file cursor should be 0 and the first line should be read.
         line = llf.readline()
-        self.assertEqual(line, self.first_line)
         llf.close()
+        self.assertEqual(line, self.first_line)
 
         # Check whether position is updated in status file.
         pos = self.read_position_in_status_file(self.ok_status_filename)
@@ -108,40 +129,167 @@ class LiveLogFileTest(unittest.TestCase):
 
         # Log file cursor should be 0 and the first line should be read.
         line = llf.readline()
+        llf.close()
         self.assertEqual(line, self.first_line)
+
+        # Check whether position is updated in status file.
+        pos = self.read_position_in_status_file(self.ok_status_filename)
+        self.assertEqual(pos, len(self.first_line))
+
+    def test_read_from_beggining(self):
+        """Read from beginning of the log file.
+        Status file is present.
+        """
+        f = open(self.ok_status_filename, "wt")
+        f.write("0\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 0 and the first line should be read.
+        line = llf.readline()
         llf.close()
+        self.assertEqual(line, self.first_line)
 
         # Check whether position is updated in status file.
         pos = self.read_position_in_status_file(self.ok_status_filename)
         self.assertEqual(pos, len(self.first_line))
 
+    def test_read_from_middle(self):
+        """Read the third line of the log file.
+        Status file content is 14 (the log file cursor position for
+        reading third line).
+        """
+        f = open(self.ok_status_filename, "wt")
+        f.write(str(len(self.first_line) + len(self.second_line)) + "\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 14 and the third line should be read.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, self.third_line)
+
+        # Check whether position is updated in status file.
+        pos = self.read_position_in_status_file(self.ok_status_filename)
+        self.assertEqual(pos, len(self.first_line) +
+            len(self.second_line) + len(self.third_line))
+
     def test_no_newline_at_end_of_status_file(self):
         """Status file contains invalid content (no string)."""
-        self.assertEqual(True, False)
+        f = open(self.ok_status_filename, "wt")
+        f.write("0")
+        f.close()
 
-    def test_read_one_line_from_log_file(self):
-        """Read a single line from the log file.
-        Status information should be updated accordingly.
-        """
-        self.assertEqual(True, False)
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 0 and the first line should be read.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, self.first_line)
+
+        # Check whether position is updated in status file.
+        pos = self.read_position_in_status_file(self.ok_status_filename)
+        self.assertEqual(pos, len(self.first_line))
 
     def test_read_two_lines_from_log_file(self):
         """Read two lines from the log file.
         Status information should be updated accordingly.
         """
-        self.assertEqual(True, False)
+        f = open(self.ok_status_filename, "wt")
+        f.write("0\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 0 and the first two lines should be read.
+        line1 = llf.readline()
+        line2 = llf.readline()
+        llf.close()
+        self.assertEqual(line1, self.first_line)
+        self.assertEqual(line2, self.second_line)
+
+        # Check whether position is updated in status file.
+        pos = self.read_position_in_status_file(self.ok_status_filename)
+        self.assertEqual(pos, len(self.first_line) + len(self.second_line))
+
+    def test_status_position_is_greater_than_log_file_length(self):
+        """Information in status file is a too greater position.
+        No data may be read. Information stays the same in status file.
+        """
+        f = open(self.ok_status_filename, "wt")
+        f.write("200\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 0 and the first line should be read.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, "")
+
+        # Position should still be 200.
+        pos = self.read_position_in_status_file(self.ok_status_filename)
+        self.assertEqual(pos, 200)
 
     def test_empty_log_file(self):
         """Log file is empty (zero length)."""
-        self.assertEqual(True, False)
+
+        f = open(self.empty_status_filename, "wt")
+        f.write("0\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.empty_log_filename)
+
+        # Log file cursor should be 0 and a blank line should be returned.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, "")
+
+        # Position should still be 0.
+        pos = self.read_position_in_status_file(self.empty_status_filename)
+        self.assertEqual(pos, 0)
 
     def test_incomplete_log_file(self):
         """Log file's last doesn't end in a new line."""
-        self.assertEqual(True, False)
+
+        f = open(self.incomplete_status_filename, "wt")
+        f.write("0\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.incomplete_log_filename)
+
+        # Log file cursor should be 0 and a blank line should be returned.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, "")
+
+        # Position should still be 0.
+        pos = self.read_position_in_status_file(self.incomplete_status_filename)
+        self.assertEqual(pos, 0)
 
     def test_readline_fails_after_close(self):
         """Use readline() after close(). Should fail."""
-        self.assertEqual(True, False)
+
+        f = open(self.ok_status_filename, "wt")
+        f.write("0\n")
+        f.close()
+
+        llf = util.LiveLogFile(self.ok_log_filename)
+
+        # Log file cursor should be 0 and the first line should be read.
+        line = llf.readline()
+        llf.close()
+        self.assertEqual(line, self.first_line)
+
+        exception_raised = False
+        try:
+            llf.readline()
+        except (IOError, ValueError), e:
+            exception_raised = True
+
+        self.assertEqual(exception_raised, True)
 
 if __name__ == "__main__":
     unittest.main()