import sys
import parsing
+import storage
class LibtorrentLogParserTest(unittest.TestCase):
"""Test suite for LibtorrentLogParser class in parsing.py."""
f = open(self.ok_bogus_file_path3, "w")
f.close()
+ # Initialize message type counters.
+ self.statmsg_cnt = 0
+ self.pstatmsg_cnt = 0
+ self.verbmsg_cnt = 0
+
# Remove no_dir in case it exists.
try:
shutil.rmtree(self.no_dir)
except OSError, e:
pass
+ def add_bogus_line(self, filename):
+ """Add bogus (non-relevant) line to file."""
+ pass
+
+ def add_status_line(self, filename):
+ """Add status line to file."""
+ pass
+
+ def add_peer_status_line(self, filename):
+ """Add peer status line to file."""
+ pass
+
+ def add_verbose_line(self, filename):
+ """Add verbose line to file."""
+ pass
+
+ def add_message_to_count(self, msg):
+ """Update message type counters according to msg."""
+ if msg.__class__ == storage.StatusMessage:
+ self.statmsg_count = self.statmsg_count + 1
+ elif msg.__class__ == storage.StatusMessage:
+ self.statmsg_count = self.statmsg_count + 1
+ elif msg.__class__ == storage.StatusMessage:
+ self.statmsg_count = self.statmsg_count + 1
+
def test_non_existent_folder(self):
"""libtorrent log folder doesn't exist.
Exception should be raised.
"""No verbose log file is present.
Use get_to_parse_list() to get files to be parsed.
"""
- # Remove status log file.
+ # Remove verbose log files.
os.remove(self.ok_verbose_file_path1)
os.remove(self.ok_verbose_file_path2)
os.remove(self.ok_verbose_file_path3)
"""Set priority as "status".
get_current_parsing_filename() should return status log file.
"""
- self.assertEqual(True, False)
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="status")
+ current = p.get_current_parsing_filename()
+
+ value = (current == self.ok_status_file_path)
+ self.assertEqual(value, True)
def test_get_parsing_verbose_priority(self):
"""Set priority as "status".
get_current_parsing_filename() should return a verbose log file.
"""
- self.assertEqual(True, False)
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ current = p.get_current_parsing_filename()
+
+ value = (current == self.ok_verbose_file_path1 or
+ current == self.ok_verbose_file_path2 or
+ current == self.ok_verbose_file_path3)
+ self.assertEqual(value, True)
def test_get_message_no_status_log_file_empty_verbose_log_files(self):
"""Status log file does not exist, verbose log files are empty.
Returned message should be None.
"""
- self.assertEqual(True, False)
+ # Remove status log file.
+ os.remove(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir)
+ msg = p.get_next_message()
+
+ self.assertEqual(msg, None)
def test_get_message_no_verbose_log_files_empty_status_log_file(self):
"""Verbose log files do not exist, status log file is empty.
Returned message should be None.
"""
- self.assertEqual(True, False)
+ # Remove verbose log files.
+ os.remove(self.ok_verbose_file_path1)
+ os.remove(self.ok_verbose_file_path2)
+ os.remove(self.ok_verbose_file_path3)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir)
+ msg = p.get_next_message()
+
+ self.assertEqual(msg, None)
def test_get_message_empty_log_files(self):
"""Log files are empty.
Returned message should be None.
"""
- self.assertEqual(True, False)
+ p = parsing.LibtorrentLogParser(self.ok_dir)
+ msg = p.get_next_message()
- def test_get_message_in_status_log_file(self):
+ self.assertEqual(msg, None)
+
+ def test_get_none_message_in_status_log_file(self):
"""Status log file consists of bogus information (no messages).
Set priority as "status".
Message is returned from verbose log file.
"""
- self.assertEqual(True, False)
+ self.add_bogus_line(self.ok_status_file_path)
+ self.add_verbose_line(self.ok_verbose_file_path1)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="status")
+ msg = p.get_next_message()
+
+ self.add_message_to_count(msg)
+
+ self.assertEqual(self.statmsg_cnt, 1)
def test_get_none_message_in_verbose_log_files(self):
"""Verbose log files consist of bogus information (no messages).
Set priority as "verbose".
Message is returned from status log file.
"""
- self.assertEqual(True, False)
+ self.add_bogus_line(self.ok_verbose_file_path1)
+ self.add_bogus_line(self.ok_verbose_file_path2)
+ self.add_bogus_line(self.ok_verbose_file_path3)
+ self.add_status_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg = p.get_next_message()
+
+ self.add_message_to_count(msg)
+
+ self.assertEqual(self.verbmsg_cnt, 1)
def test_get_status_message(self):
"""Return a status message from status log file."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg = p.get_next_message()
+
+ self.add_message_to_count(msg)
+
+ self.assertEqual(self.statmsg_cnt, 1)
def test_get_status_messages(self):
"""Return multiple status messages from status log file."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_status_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.statmsg_cnt, 2)
def test_get_status_messages_different_files(self):
"""Return multiple status messages from different log files."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_status_line(self.ok_verbose_file_path1)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.statmsg_cnt, 2)
def test_get_peer_status_message(self):
"""Return a peer status message from status log file."""
- self.assertEqual(True, False)
+ self.add_peer_status_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg = p.get_next_message()
+
+ self.add_message_to_count(msg)
+
+ self.assertEqual(self.pstatmsg_cnt, 1)
def test_get_peer_status_messages(self):
"""Return multiple peer status messages from status log file."""
- self.assertEqual(True, False)
+ self.add_peer_status_line(self.ok_status_file_path)
+ self.add_peer_status_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.pstatmsg_cnt, 2)
def test_get_peer_status_messages_different_files(self):
"""Return multiple peer status messages from different log files."""
- self.assertEqual(True, False)
+ self.add_peer_status_line(self.ok_status_file_path)
+ self.add_peer_status_line(self.ok_verbose_file_path1)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.pstatmsg_cnt, 2)
def test_get_verbose_message(self):
"""Return a verbose message from verbose log file."""
- self.assertEqual(True, False)
+ self.add_verbose_line(self.ok_verbose_file_path1)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="status")
+ msg = p.get_next_message()
+
+ self.add_message_to_count(msg)
+
+ self.assertEqual(self.verbmsg_cnt, 1)
def test_get_verbose_messages(self):
"""Return multiple verbose messages from verbose log file."""
- self.assertEqual(True, False)
+ self.add_verbose_line(self.ok_verbose_file_path1)
+ self.add_verbose_line(self.ok_verbose_file_path1)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.verbmsg_cnt, 2)
def test_get_verbose_messages_different_files(self):
"""Return multiple verbose messages from different log files."""
- self.assertEqual(True, False)
+ self.add_verbose_line(self.ok_verbose_file_path1)
+ self.add_verbose_line(self.ok_verbose_file_path2)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+
+ self.assertEqual(self.verbmsg_cnt, 2)
def test_get_mixed_messages_same_file(self):
"""Status file contains multiple messages of different types."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_peer_status_line(self.ok_status_file_path)
+ self.add_verbose_line(self.ok_status_file_path)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+ msg3 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+ self.add_message_to_count(msg3)
+
+ value = (self.statmsg_cnt == 1 and self.pstatmsg_cnt == 1 and \
+ self.verbmsg_cnt == 1)
+ self.assertEqual(value, True)
def test_get_mixed_messages_own_file(self):
"""Each log file contains a single type of messages."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_peer_status_line(self.ok_verbose_file_path1)
+ self.add_peer_status_line(self.ok_verbose_file_path1)
+ self.add_verbose_line(self.ok_verbose_file_path2)
+ self.add_verbose_line(self.ok_verbose_file_path2)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+ msg3 = p.get_next_message()
+ msg4 = p.get_next_message()
+ msg5 = p.get_next_message()
+ msg6 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+ self.add_message_to_count(msg3)
+ self.add_message_to_count(msg4)
+ self.add_message_to_count(msg5)
+ self.add_message_to_count(msg6)
+
+ value = (self.statmsg_cnt == 2 and self.pstatmsg_cnt == 2 and \
+ self.verbmsg_cnt == 2)
+ self.assertEqual(value, True)
def test_get_mixed_messages_different_files(self):
"""Log files contain different message types."""
- self.assertEqual(True, False)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_status_line(self.ok_status_file_path)
+ self.add_peer_status_line(self.ok_verbose_file_path1)
+ self.add_peer_status_line(self.ok_verbose_file_path1)
+ self.add_verbose_line(self.ok_verbose_file_path2)
+ self.add_verbose_line(self.ok_verbose_file_path2)
+
+ p = parsing.LibtorrentLogParser(self.ok_dir, priority="verbose")
+ msg1 = p.get_next_message()
+ msg2 = p.get_next_message()
+ msg3 = p.get_next_message()
+ msg4 = p.get_next_message()
+ msg5 = p.get_next_message()
+ msg6 = p.get_next_message()
+
+ self.add_message_to_count(msg1)
+ self.add_message_to_count(msg2)
+ self.add_message_to_count(msg3)
+ self.add_message_to_count(msg4)
+ self.add_message_to_count(msg5)
+ self.add_message_to_count(msg6)
+
+ value = (self.statmsg_cnt == 2 and self.pstatmsg_cnt == 2 and \
+ self.verbmsg_cnt == 2)
+ self.assertEqual(value, True)
class TriblerLogParserTest(unittest.TestCase):
projects / cs-p2p-next.git / blobdiff