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