ppf: Add tests for add_*_message.

diff --git a/ppf/new/tests/test_storage.py b/ppf/new/tests/test_storage.py
index 9766e63..756b903 100644 (file)
--- a/ppf/new/tests/test_storage.py
+++ b/ppf/new/tests/test_storage.py
@@ -145,14 +145,13 @@ class TreeTextFileAccessTest(unittest.TestCase):
         shutil.rmtree(self.path)
 
     def test_add_swarm_folder_and_file_are_created(self):
-        expected_swarm_subfolder = 1
-
         # Create instance of class and add swarm.
         a = storage.TreeTextFileAccess(self.path)
         s = storage.Swarm(torrent_filename="fedora.torrent",
                 data_size=102400)
         a.add_swarm(s)
 
+        expected_swarm_subfolder = 1
         expected_swarm_path = os.path.join(self.path,
                 str(expected_swarm_subfolder))
         expected_swarm_config = os.path.join(expected_swarm_path, "swarm.conf")
@@ -162,9 +161,6 @@ class TreeTextFileAccessTest(unittest.TestCase):
         self.assertEqual(expected_true, True)
 
     def test_add_client_session_folder_and_file_are_created(self):
-        expected_swarm_subfolder = 1
-        expected_session_subfolder = 1
-
         a = storage.TreeTextFileAccess(self.path)
 
         # Add swarm.
@@ -179,6 +175,8 @@ class TreeTextFileAccessTest(unittest.TestCase):
                 public_port="50500", ds_limit=300, us_limit=200)
         a.add_client_session(cs)
 
+        expected_swarm_subfolder = 1
+        expected_session_subfolder = 1
         expected_swarm_path = os.path.join(self.path,
                 str(expected_swarm_subfolder))
         expected_session_path = os.path.join(expected_swarm_path,
@@ -191,19 +189,106 @@ class TreeTextFileAccessTest(unittest.TestCase):
         self.assertEqual(expected_true, True)
 
     def test_add_peer_status_message_file_is_created(self):
-        self.assertEqual(True, True)
+        a = storage.TreeTextFileAccess(self.path)
+
+        # Add swarm.
+        s = storage.Swarm(torrent_filename="fedora.torrent",
+                data_size=102400)
+        a.add_swarm(s)
+
+        # Add client session.
+        cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
+                system_os="Linux", system_os_version="2.6.26",
+                system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
+                public_port="50500", ds_limit=300, us_limit=200)
+        a.add_client_session(cs)
+
+        # Add peer status message.
+        msg = storage.PeerStatusMessage(swarm_id=1, client_session_id=1,
+                peer_ip="141.85.224.202", peer_port="12345",
+                download_speed=13, upload_speed=98)
+        a.add_peer_status_message(msg)
+
+        expected_swarm_subfolder = 1
+        expected_session_subfolder = 1
+        expected_swarm_path = os.path.join(self.path,
+                str(expected_swarm_subfolder))
+        expected_session_path = os.path.join(expected_swarm_path,
+                str(expected_session_subfolder))
+        expected_message_file = os.path.join(expected_session_path,
+                "peer_status.txt")
+
+        self.assertEqual(os.path.isfile(expected_message_file), True)
 
     def test_add_peer_status_message_number_of_lines(self):
         self.assertEqual(True, True)
 
     def test_add_status_message_file_is_created(self):
-        self.assertEqual(True, True)
+        a = storage.TreeTextFileAccess(self.path)
+
+        # Add swarm.
+        s = storage.Swarm(torrent_filename="fedora.torrent",
+                data_size=102400)
+        a.add_swarm(s)
+
+        # Add client session.
+        cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
+                system_os="Linux", system_os_version="2.6.26",
+                system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
+                public_port="50500", ds_limit=300, us_limit=200)
+        a.add_client_session(cs)
+
+        # Add peer status message.
+        msg = storage.StatusMessage(swarm_id=1, client_session_id=1,
+                num_peers=10, num_dht_peers=3, download_speed=102,
+                upload_speed=99, download_size=10213, upload_size=3301)
+        a.add_status_message(msg)
+
+        expected_swarm_subfolder = 1
+        expected_session_subfolder = 1
+        expected_swarm_path = os.path.join(self.path,
+                str(expected_swarm_subfolder))
+        expected_session_path = os.path.join(expected_swarm_path,
+                str(expected_session_subfolder))
+        expected_message_file = os.path.join(expected_session_path,
+                "status.txt")
+
+        self.assertEqual(os.path.isfile(expected_message_file), True)
 
     def test_add_status_message_number_of_lines(self):
         self.assertEqual(True, True)
 
     def test_add_verbose_message_file_is_created(self):
-        self.assertEqual(True, True)
+        a = storage.TreeTextFileAccess(self.path)
+
+        # Add swarm.
+        s = storage.Swarm(torrent_filename="fedora.torrent",
+                data_size=102400)
+        a.add_swarm(s)
+
+        # Add client session.
+        cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
+                system_os="Linux", system_os_version="2.6.26",
+                system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
+                public_port="50500", ds_limit=300, us_limit=200)
+        a.add_client_session(cs)
+
+        # Add peer status message.
+        msg = storage.VerboseMessage(swarm_id=1, client_session_id=1,
+            transfer_direction="to", peer_ip="141.85.224.202",
+            peer_port="12345", message_type="CHOKE")
+        a.add_verbose_message(msg)
+
+        expected_swarm_subfolder = 1
+        expected_session_subfolder = 1
+        expected_swarm_path = os.path.join(self.path,
+                str(expected_swarm_subfolder))
+        expected_session_path = os.path.join(expected_swarm_path,
+                str(expected_session_subfolder))
+        expected_message_file = os.path.join(expected_session_path,
+                "verbose.txt")
+
+        self.assertEqual(os.path.isfile(expected_message_file), True)
 
     def test_add_verbose_message_number_of_lines(self):
         self.assertEqual(True, True)