ppf/new: Complete tests for util.py.
[cs-p2p-next.git] / ppf / new / tests / test_util.py
1 """
2 Test suite for util.py. Uses unittest module.
3
4 2011, Razvan Deaconescu, razvan.deaconescu@cs.pub.ro.
5 """
6
7 import unittest
8 import os
9 import os.path
10 import shutil
11 import subprocess
12 import sys
13
14 import util
15
16 class LiveLogStatusTest(unittest.TestCase):
17     """Test suite for LiveLogStatus class in util.py."""
18
19     def test_init(self):
20         """Class constructor works properly."""
21         status = util.LiveLogStatus(12345)
22
23         self.assertEqual(status.position, 12345)
24
25 class LiveLogFileTest(unittest.TestCase):
26     """Test suite for LiveLogFile class in util.py."""
27     ok_log_filename = "ok.log"
28     ok_status_filename = "ok.log.status"
29     empty_log_filename = "empty.log"
30     empty_status_filename = "empty.log.status"
31     incomplete_log_filename = "incomplete.log"
32     incomplete_status_filename = "incomplete.log.status"
33     no_log_filename = "no.log"
34     first_line = "line 1\n"
35     second_line = "line 1\n"
36     third_line = "line 1\n"
37
38     def remove_status_files(self):
39         try:
40             os.remove(self.ok_status_filename)
41         except OSError, e:
42             pass
43
44         try:
45             os.remove(self.empty_status_filename)
46         except OSError, e:
47             pass
48
49         try:
50             os.remove(self.incomplete_status_filename)
51         except OSError, e:
52             pass
53
54     def setUp(self):
55         """Create test log files."""
56         f = open(self.ok_log_filename, "wt")
57         f.write(self.first_line)
58         f.write(self.second_line)
59         f.write(self.third_line)
60         f.close()
61
62         f = open(self.empty_log_filename, "wt")
63         f.close()
64
65         f = open(self.incomplete_log_filename, "wt")
66         f.write("no newline")
67         f.close()
68
69         # Remove "no log" file and status files in case they exist.
70         try:
71             os.remove(self.no_log_filename)
72         except OSError, e:
73             pass
74         self.remove_status_files()
75
76     def tearDown(self):
77         """Remove test log files."""
78         os.remove(self.ok_log_filename)
79         os.remove(self.empty_log_filename)
80         os.remove(self.incomplete_log_filename)
81
82         # Remove "no log" file and status files in case they exist.
83         try:
84             os.remove(self.no_log_filename)
85         except OSError, e:
86             pass
87         self.remove_status_files()
88
89     def read_position_in_status_file(self, status_filename):
90         """Read first line in status file as integer."""
91         f = open(status_filename, "rt")
92         line = f.readline()
93         return int(line)
94
95     def test_no_log_file(self):
96         """No log file is present."""
97         exception_raised = False
98         try:
99             llf = util.LiveLogFile(self.no_log_filename)
100         except IOError, e:
101             exception_raised = True
102
103         self.assertEqual(exception_raised, True)
104
105     def test_no_status_file(self):
106         """No status file is present."""
107
108         # Status file should be created by constructor.
109         llf = util.LiveLogFile(self.ok_log_filename)
110         self.assertEqual(os.access(self.ok_status_filename, os.W_OK), True)
111
112         # Log file cursor should be 0 and the first line should be read.
113         line = llf.readline()
114         llf.close()
115         self.assertEqual(line, self.first_line)
116
117         # Check whether position is updated in status file.
118         pos = self.read_position_in_status_file(self.ok_status_filename)
119         self.assertEqual(pos, len(self.first_line))
120
121     def test_invalid_content_in_status_file(self):
122         """Status file contains invalid content (no string)."""
123
124         f = open(self.ok_status_filename, "wt")
125         f.write("ana-are-mere\n")
126         f.close()
127
128         llf = util.LiveLogFile(self.ok_log_filename)
129
130         # Log file cursor should be 0 and the first line should be read.
131         line = llf.readline()
132         llf.close()
133         self.assertEqual(line, self.first_line)
134
135         # Check whether position is updated in status file.
136         pos = self.read_position_in_status_file(self.ok_status_filename)
137         self.assertEqual(pos, len(self.first_line))
138
139     def test_read_from_beggining(self):
140         """Read from beginning of the log file.
141         Status file is present.
142         """
143         f = open(self.ok_status_filename, "wt")
144         f.write("0\n")
145         f.close()
146
147         llf = util.LiveLogFile(self.ok_log_filename)
148
149         # Log file cursor should be 0 and the first line should be read.
150         line = llf.readline()
151         llf.close()
152         self.assertEqual(line, self.first_line)
153
154         # Check whether position is updated in status file.
155         pos = self.read_position_in_status_file(self.ok_status_filename)
156         self.assertEqual(pos, len(self.first_line))
157
158     def test_read_from_middle(self):
159         """Read the third line of the log file.
160         Status file content is 14 (the log file cursor position for
161         reading third line).
162         """
163         f = open(self.ok_status_filename, "wt")
164         f.write(str(len(self.first_line) + len(self.second_line)) + "\n")
165         f.close()
166
167         llf = util.LiveLogFile(self.ok_log_filename)
168
169         # Log file cursor should be 14 and the third line should be read.
170         line = llf.readline()
171         llf.close()
172         self.assertEqual(line, self.third_line)
173
174         # Check whether position is updated in status file.
175         pos = self.read_position_in_status_file(self.ok_status_filename)
176         self.assertEqual(pos, len(self.first_line) +
177             len(self.second_line) + len(self.third_line))
178
179     def test_no_newline_at_end_of_status_file(self):
180         """Status file contains invalid content (no string)."""
181         f = open(self.ok_status_filename, "wt")
182         f.write("0")
183         f.close()
184
185         llf = util.LiveLogFile(self.ok_log_filename)
186
187         # Log file cursor should be 0 and the first line should be read.
188         line = llf.readline()
189         llf.close()
190         self.assertEqual(line, self.first_line)
191
192         # Check whether position is updated in status file.
193         pos = self.read_position_in_status_file(self.ok_status_filename)
194         self.assertEqual(pos, len(self.first_line))
195
196     def test_read_two_lines_from_log_file(self):
197         """Read two lines from the log file.
198         Status information should be updated accordingly.
199         """
200         f = open(self.ok_status_filename, "wt")
201         f.write("0\n")
202         f.close()
203
204         llf = util.LiveLogFile(self.ok_log_filename)
205
206         # Log file cursor should be 0 and the first two lines should be read.
207         line1 = llf.readline()
208         line2 = llf.readline()
209         llf.close()
210         self.assertEqual(line1, self.first_line)
211         self.assertEqual(line2, self.second_line)
212
213         # Check whether position is updated in status file.
214         pos = self.read_position_in_status_file(self.ok_status_filename)
215         self.assertEqual(pos, len(self.first_line) + len(self.second_line))
216
217     def test_status_position_is_greater_than_log_file_length(self):
218         """Information in status file is a too greater position.
219         No data may be read. Information stays the same in status file.
220         """
221         f = open(self.ok_status_filename, "wt")
222         f.write("200\n")
223         f.close()
224
225         llf = util.LiveLogFile(self.ok_log_filename)
226
227         # Log file cursor should be 0 and the first line should be read.
228         line = llf.readline()
229         llf.close()
230         self.assertEqual(line, "")
231
232         # Position should still be 200.
233         pos = self.read_position_in_status_file(self.ok_status_filename)
234         self.assertEqual(pos, 200)
235
236     def test_empty_log_file(self):
237         """Log file is empty (zero length)."""
238
239         f = open(self.empty_status_filename, "wt")
240         f.write("0\n")
241         f.close()
242
243         llf = util.LiveLogFile(self.empty_log_filename)
244
245         # Log file cursor should be 0 and a blank line should be returned.
246         line = llf.readline()
247         llf.close()
248         self.assertEqual(line, "")
249
250         # Position should still be 0.
251         pos = self.read_position_in_status_file(self.empty_status_filename)
252         self.assertEqual(pos, 0)
253
254     def test_incomplete_log_file(self):
255         """Log file's last doesn't end in a new line."""
256
257         f = open(self.incomplete_status_filename, "wt")
258         f.write("0\n")
259         f.close()
260
261         llf = util.LiveLogFile(self.incomplete_log_filename)
262
263         # Log file cursor should be 0 and a blank line should be returned.
264         line = llf.readline()
265         llf.close()
266         self.assertEqual(line, "")
267
268         # Position should still be 0.
269         pos = self.read_position_in_status_file(self.incomplete_status_filename)
270         self.assertEqual(pos, 0)
271
272     def test_readline_fails_after_close(self):
273         """Use readline() after close(). Should fail."""
274
275         f = open(self.ok_status_filename, "wt")
276         f.write("0\n")
277         f.close()
278
279         llf = util.LiveLogFile(self.ok_log_filename)
280
281         # Log file cursor should be 0 and the first line should be read.
282         line = llf.readline()
283         llf.close()
284         self.assertEqual(line, self.first_line)
285
286         exception_raised = False
287         try:
288             llf.readline()
289         except (IOError, ValueError), e:
290             exception_raised = True
291
292         self.assertEqual(exception_raised, True)
293
294 if __name__ == "__main__":
295     unittest.main()