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