ppf/new/tests/test_storage.py

   1 """
   2 Test suite for storage. 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 sqlite3
  12 import subprocess
  13 import sys
  14
  15 import storage
  16
  17 def create_non_numeric_subfolders(path):
  18     os.mkdir(os.path.join(path, "ana"))
  19     os.mkdir(os.path.join(path, "are"))
  20     os.mkdir(os.path.join(path, "mere"))
  21     os.mkdir(os.path.join(path, "si"))
  22     os.mkdir(os.path.join(path, "pere"))
  23
  24 def create_numeric_subfolders_sequential(path):
  25     os.mkdir(os.path.join(path, "1"))
  26     os.mkdir(os.path.join(path, "2"))
  27     os.mkdir(os.path.join(path, "3"))
  28     os.mkdir(os.path.join(path, "4"))
  29     os.mkdir(os.path.join(path, "5"))
  30
  31 def create_numeric_subfolders_non_sequential(path):
  32     os.mkdir(os.path.join(path, "32"))
  33     os.mkdir(os.path.join(path, "5"))
  34     os.mkdir(os.path.join(path, "423"))
  35     os.mkdir(os.path.join(path, "1401"))
  36     os.mkdir(os.path.join(path, "92"))
  37
  38 def create_numeric_files(path):
  39     f = open(os.path.join(path, "33"), 'w')
  40     f.close()
  41     f = open(os.path.join(path, "6"), 'w')
  42     f.close()
  43     f = open(os.path.join(path, "424"), 'w')
  44     f.close()
  45     f = open(os.path.join(path, "1402"), 'w')
  46     f.close()
  47     f = open(os.path.join(path, "93"), 'w')
  48     f.close()
  49
  50 class StorageTest(unittest.TestCase):
  51     """
  52     Test suite for stand alone functions in storage.py.
  53     """
  54
  55     # Class specific variables. Initialized in setUp, used throughout tests.
  56     path = "/tmp/ttfa-test"
  57
  58     def setUp(self):
  59         """Create folder."""
  60         os.mkdir(self.path)
  61
  62     def tearDown(self):
  63         """Remove base folder."""
  64         shutil.rmtree(self.path)
  65
  66     def test_find_last_numeric_subfolder_no_subfolders(self):
  67         """Test return of None when no subfolders are present."""
  68         id = storage.find_last_numeric_subfolder(self.path)
  69         self.assertEqual(id, None)
  70
  71     def test_find_last_numeric_subfolder_no_numeric_subfolders(self):
  72         """Test return of None when no numeric subfolders are present."""
  73         create_non_numeric_subfolders(self.path)
  74
  75         id = storage.find_last_numeric_subfolder(self.path)
  76         self.assertEqual(id, None)
  77
  78     def test_find_last_numeric_subfolder_only_numeric_subfolders_sequential(self):
  79         """
  80         Test return of correct id when sequential numeric named
  81         subfolders only are present (1, 2, 3, ...).
  82         """
  83         create_numeric_subfolders_sequential(self.path)
  84
  85         id = storage.find_last_numeric_subfolder(self.path)
  86         self.assertEqual(id, 5)
  87
  88     def test_find_last_numeric_subfolder_only_numeric_subfolders_nonsequential(self):
  89         """
  90         Test return of correct id when nonsequential numeric named
  91         subfolders only are present (32, 5, 423, ...).
  92         """
  93         create_numeric_subfolders_non_sequential(self.path)
  94
  95         id = storage.find_last_numeric_subfolder(self.path)
  96         self.assertEqual(id, 1401)
  97
  98     def test_find_last_numeric_subfolder_mixed_subfolders_sequential(self):
  99         """
 100         Test return of None when mixed named subfolders (numeric and
 101         others) are present. Numeric folders are sequential.
 102         """
 103         create_non_numeric_subfolders(self.path)
 104         create_numeric_subfolders_sequential(self.path)
 105
 106         id = storage.find_last_numeric_subfolder(self.path)
 107         self.assertEqual(id, 5)
 108
 109     def test_find_last_numeric_subfolder_mixed_subfolders_nonsequential(self):
 110         """
 111         Test return of None when mixed named subfolders (numeric and
 112         others) are present. Numeric folders are non-sequential.
 113         """
 114         create_non_numeric_subfolders(self.path)
 115         create_numeric_subfolders_non_sequential(self.path)
 116
 117         id = storage.find_last_numeric_subfolder(self.path)
 118         self.assertEqual(id, 1401)
 119
 120     def test_find_last_numeric_subfolder_mixed_subfolders_files(self):
 121         """
 122         Test return of None when mixed named subfolders (numeric and
 123         others) together with files are present. Numeric folders are
 124         non-sequential.
 125         """
 126         create_non_numeric_subfolders(self.path)
 127         create_numeric_subfolders_non_sequential(self.path)
 128         create_numeric_files(self.path)
 129
 130         id = storage.find_last_numeric_subfolder(self.path)
 131         self.assertEqual(id, 1401)
 132
 133
 134 class TreeTextFileAccessTest(unittest.TestCase):
 135     """
 136     Test suite for TreeTextFileAccess class in storage.py.
 137     """
 138
 139     # Class specific variables. Initialized in setUp, used throughout tests.
 140     path = "/tmp/ttfa-test"
 141
 142     def setUp(self):
 143         """Create folder. Initialize object. """
 144         os.mkdir(self.path)
 145
 146     def tearDown(self):
 147         """Remove base folder."""
 148         shutil.rmtree(self.path)
 149
 150     def test_add_swarm_folder_and_file_are_created(self):
 151         # Create instance of class and add swarm.
 152         a = storage.TreeTextFileAccess(self.path)
 153         s = storage.Swarm(torrent_filename="fedora.torrent",
 154                 data_size=102400)
 155         a.add_swarm(s)
 156
 157         expected_swarm_subfolder = 1
 158         expected_swarm_path = os.path.join(self.path,
 159                 str(expected_swarm_subfolder))
 160         expected_swarm_config = os.path.join(expected_swarm_path, "swarm.conf")
 161
 162         self.assertEqual(os.path.isdir(expected_swarm_path), True)
 163         self.assertEqual(os.path.isfile(expected_swarm_config), True)
 164
 165     def test_add_client_session_folder_and_file_are_created(self):
 166         a = storage.TreeTextFileAccess(self.path)
 167
 168         # Add swarm.
 169         s = storage.Swarm(torrent_filename="fedora.torrent",
 170                 data_size=102400)
 171         a.add_swarm(s)
 172
 173         # Add client session.
 174         cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
 175                 system_os="Linux", system_os_version="2.6.26",
 176                 system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
 177                 public_port="50500", ds_limit=300, us_limit=200)
 178         a.add_client_session(cs)
 179
 180         expected_swarm_subfolder = 1
 181         expected_session_subfolder = 1
 182         expected_swarm_path = os.path.join(self.path,
 183                 str(expected_swarm_subfolder))
 184         expected_session_path = os.path.join(expected_swarm_path,
 185                 str(expected_session_subfolder))
 186         expected_session_config = os.path.join(expected_session_path,
 187                 "client_session.conf")
 188
 189         self.assertEqual(os.path.isdir(expected_session_path), True)
 190         self.assertEqual(os.path.isfile(expected_session_config), True)
 191
 192     def test_add_peer_status_message_file_is_created(self):
 193         a = storage.TreeTextFileAccess(self.path)
 194
 195         # Add swarm.
 196         s = storage.Swarm(torrent_filename="fedora.torrent",
 197                 data_size=102400)
 198         a.add_swarm(s)
 199
 200         # Add client session.
 201         cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
 202                 system_os="Linux", system_os_version="2.6.26",
 203                 system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
 204                 public_port="50500", ds_limit=300, us_limit=200)
 205         a.add_client_session(cs)
 206
 207         # Add peer status message.
 208         msg = storage.PeerStatusMessage(swarm_id=1, client_session_id=1,
 209                 peer_ip="141.85.224.202", peer_port="12345",
 210                 download_speed=13, upload_speed=98)
 211         a.add_peer_status_message(msg)
 212
 213         expected_swarm_subfolder = 1
 214         expected_session_subfolder = 1
 215         expected_swarm_path = os.path.join(self.path,
 216                 str(expected_swarm_subfolder))
 217         expected_session_path = os.path.join(expected_swarm_path,
 218                 str(expected_session_subfolder))
 219         expected_message_file = os.path.join(expected_session_path,
 220                 "peer_status.txt")
 221
 222         self.assertEqual(os.path.isfile(expected_message_file), True)
 223
 224     def test_add_peer_status_message_number_of_lines(self):
 225         self.assertEqual(True, True)
 226
 227     def test_add_status_message_file_is_created(self):
 228         a = storage.TreeTextFileAccess(self.path)
 229
 230         # Add swarm.
 231         s = storage.Swarm(torrent_filename="fedora.torrent",
 232                 data_size=102400)
 233         a.add_swarm(s)
 234
 235         # Add client session.
 236         cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
 237                 system_os="Linux", system_os_version="2.6.26",
 238                 system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
 239                 public_port="50500", ds_limit=300, us_limit=200)
 240         a.add_client_session(cs)
 241
 242         # Add peer status message.
 243         msg = storage.StatusMessage(swarm_id=1, client_session_id=1,
 244                 num_peers=10, num_dht_peers=3, download_speed=102,
 245                 upload_speed=99, download_size=10213, upload_size=3301)
 246         a.add_status_message(msg)
 247
 248         expected_swarm_subfolder = 1
 249         expected_session_subfolder = 1
 250         expected_swarm_path = os.path.join(self.path,
 251                 str(expected_swarm_subfolder))
 252         expected_session_path = os.path.join(expected_swarm_path,
 253                 str(expected_session_subfolder))
 254         expected_message_file = os.path.join(expected_session_path,
 255                 "status.txt")
 256
 257         self.assertEqual(os.path.isfile(expected_message_file), True)
 258
 259     def test_add_status_message_number_of_lines(self):
 260         self.assertEqual(True, True)
 261
 262     def test_add_verbose_message_file_is_created(self):
 263         a = storage.TreeTextFileAccess(self.path)
 264
 265         # Add swarm.
 266         s = storage.Swarm(torrent_filename="fedora.torrent",
 267                 data_size=102400)
 268         a.add_swarm(s)
 269
 270         # Add client session.
 271         cs = storage.ClientSession(swarm_id=1, btclient="Tribler",
 272                 system_os="Linux", system_os_version="2.6.26",
 273                 system_ram=2048, system_cpu=3000, public_ip="141.85.224.201",
 274                 public_port="50500", ds_limit=300, us_limit=200)
 275         a.add_client_session(cs)
 276
 277         # Add peer status message.
 278         msg = storage.VerboseMessage(swarm_id=1, client_session_id=1,
 279             transfer_direction="send", peer_ip="141.85.224.202",
 280             peer_port="12345", message_type="CHOKE")
 281         a.add_verbose_message(msg)
 282
 283         expected_swarm_subfolder = 1
 284         expected_session_subfolder = 1
 285         expected_swarm_path = os.path.join(self.path,
 286                 str(expected_swarm_subfolder))
 287         expected_session_path = os.path.join(expected_swarm_path,
 288                 str(expected_session_subfolder))
 289         expected_message_file = os.path.join(expected_session_path,
 290                 "verbose.txt")
 291
 292         self.assertEqual(os.path.isfile(expected_message_file), True)
 293
 294     def test_add_verbose_message_number_of_lines(self):
 295         self.assertEqual(True, True)
 296
 297 class SQLiteDatabaseAccessTest(unittest.TestCase):
 298     """
 299     Test suite for SQLiteDatabaseAccess class in storage.py.
 300     """
 301
 302     # Class specific variables. Initialized in setUp, used throughout tests.
 303     database = "test.db"
 304     sql_create_script = "p2p-log-sqlite.sql"
 305     sql_init_script = "p2p-init-test.sql"
 306     expected_swarms_count = 0
 307     expected_sessions_count = 0
 308     expected_statmsg_count = 0
 309     expected_verbmsg_count = 0
 310
 311     def setUp(self):
 312         """ Create database file and instantiate objects. """
 313         # Create database file. Create tables and indices.
 314         # TODO: Check exceptions.
 315         sql_create_script_path = os.path.join(os.path.dirname(__file__),
 316                 self.sql_create_script)
 317         f = open(sql_create_script_path, 'r')
 318         p = subprocess.Popen(["sqlite3", self.database], stdin=f)
 319         ret = os.waitpid(p.pid, 0)[1]
 320         f.close()
 321
 322         # Populate database.
 323         # TODO: Check exceptions.
 324         sql_init_script_path = os.path.join(os.path.dirname(__file__),
 325                 self.sql_init_script)
 326         f = open(sql_init_script_path, 'r')
 327         p = subprocess.Popen(["sqlite3", self.database], stdin=f)
 328         ret = os.waitpid(p.pid, 0)[1]
 329         f.close()
 330
 331         # Initialize to number of table entries inserted in the init script.
 332         self.expected_swarms_count = 2
 333         self.expected_sessions_count = 2
 334         self.expected_statmsg_count = 2
 335         self.expected_verbmsg_count = 2
 336
 337     def tearDown(self):
 338         """ Close connection and remove database file. """
 339         os.remove(self.database)
 340
 341     def test_add_swarm_new_entry_in_table(self):
 342         self.assertEqual(True, False)
 343
 344     def test_add_client_session_new_entry_in_table(self):
 345         self.assertEqual(True, False)
 346
 347     def test_add_peer_status_message_new_entry_in_table(self):
 348         self.assertEqual(True, False)
 349
 350     def test_add_peer_status_message_new_entry_in_table(self):
 351         self.assertEqual(True, False)
 352
 353     def test_add_status_message_new_entry_in_table(self):
 354         self.assertEqual(True, False)
 355
 356     def test_add_verbose_message_new_entry_in_databas(self):
 357         self.assertEqual(True, False)
 358
 359 if __name__ == "__main__":
 360     unittest.main()