ppf/new/tests/test_config.py

   1 """
   2 Test suite for config. Uses unittest module.
   3
   4 2011, Mariana Marasoiu, mariana.marasoiu@gmail.com
   5 """
   6
   7 import unittest
   8 import os
   9 import os.path
  10 import datetime
  11 import sys
  12 import MySQLdb
  13 import subprocess
  14
  15 import config
  16 import storage
  17
  18 class SwarmDescriptionTest(unittest.TestCase):
  19
  20     """Test suite for SwarmDescription class in config.py."""
  21
  22     # Class specific variables. Initialized in setUp, used throughout tests.
  23     config_file = "config.sample.ini"
  24     test_data = None
  25     sd = None
  26
  27     def setUp(self):
  28         self.test_data = {'swarm': {'torrent_filename':
  29                                       'fedora-11-cds.torrent',
  30                                     'data_size': '12345678',
  31                                     'description': ('BitTorrent Distribution '
  32                                       'Experiment, Fedora 11, i386 netinst')},
  33                           'p2p-next-01': {'bittorrent_client':
  34                                             'libtorrent-rasterbar',
  35                                           'system_os': 'Linux',
  36                                           'system_os_version': '2.6.26',
  37                                           'system_ram': '1999',
  38                                           'system_cpu': '2992',
  39                                           'public_ip': '141.85.224.201',
  40                                           'public_port': '6881',
  41                                           'ds_limit': '0',
  42                                           'us_limit': '0',
  43                                           'start_time':
  44                                             'Thu Aug 11 15:36:35 EEST 2010',
  45                                           'dht_enabled': 'true',
  46                                           'pxe_enabled': 'false',
  47                                           'streaming_enabled': 'false',
  48                                           'description': 'NCIT Cluster System',
  49                                           'log_file':
  50                                             'p2p-next-01.grid.pub.ro/btex-fe'
  51                                             'dora-11/dbex-fedora-cds.tar.gz'},
  52                           'p2p-next-03': {'bittorrent_client':
  53                                               'libtorrent-rasterbar',
  54                                           'system_os': 'Linux',
  55                                           'system_os_version': '2.6.26',
  56                                           'system_ram': '2007',
  57                                           'system_cpu': '2992',
  58                                           'public_ip': '141.85.224.209',
  59                                           'public_port': '6881',
  60                                           'ds_limit': '0',
  61                                           'us_limit': '0',
  62                                           'start_time':
  63                                             'Thu Aug 11 15:36:35 EEST 2010',
  64                                           'dht_enabled': 'true',
  65                                           'pxe_enabled': 'false',
  66                                           'streaming_enabled': 'false',
  67                                           'description': 'NCIT Cluster System',
  68                                           'log_file':
  69                                             'p2p-next-03.grid.pub.ro/btex-fe'
  70                                             'dora-11/dbex-fedora-cds.tar.gz'},
  71                           'p2p-next-04': {'bittorrent_client':
  72                                             'libtorrent-rasterbar',
  73                                           'system_os': 'Linux',
  74                                           'system_os_version': '2.6.26',
  75                                           'system_ram': '2007',
  76                                           'system_cpu': '2992',
  77                                           'public_ip': '141.85.224.204',
  78                                           'public_port': '6881',
  79                                           'ds_limit': '0',
  80                                           'us_limit': '0',
  81                                           'start_time':
  82                                             'Thu Aug 11 15:36:35 EEST 2010',
  83                                           'dht_enabled': 'true',
  84                                           'pxe_enabled': 'false',
  85                                           'streaming_enabled': 'false',
  86                                           'description': 'NCIT Cluster System',
  87                                           'log_file':
  88                                             'p2p-next-04.grid.pub.ro/btex-fe'
  89                                             'dora-11/dbex-fedora-cds.tar.gz'},
  90                          'p2p-next-09': {'bittorrent_client':
  91                                             'libtorrent-rasterbar',
  92                                           'system_os': 'Linux',
  93                                           'system_os_version': '2.6.26',
  94                                           'system_ram': '1999',
  95                                           'system_cpu': '2992',
  96                                           'public_ip': '141.85.224.209',
  97                                           'public_port': '6881',
  98                                           'ds_limit': '0',
  99                                           'us_limit': '0',
 100                                           'start_time':
 101                                             'Thu Aug 11 15:36:35 EEST 2010',
 102                                           'dht_enabled': 'true',
 103                                           'pxe_enabled': 'false',
 104                                           'streaming_enabled': 'false',
 105                                           'description': 'NCIT Cluster System',
 106                                           'log_file':
 107                                             'p2p-next-09.grid.pub.ro/btex-fe'
 108                                             'dora-11/dbex-fedora-cds.tar.gz'}}
 109
 110         config_file_path = os.path.join(os.path.dirname(__file__),
 111                                         self.config_file)
 112         self.sd = config.SwarmDescription()
 113         self.sd.load(config_file_path)
 114
 115     def test_load(self):
 116         # Check for the same sections and items.
 117         # Use sort() method because self.test_data is unordered.
 118         self.assertEqual(self.sd.data.sections().sort(),
 119                          list(self.test_data.keys()).sort())
 120         for section in self.sd.data.sections():
 121             self.assertEqual(self.sd.data.items(section).sort(),
 122                              list(self.test_data[section]).sort())
 123
 124     def test_store(self):
 125         self.sd.store("session_config.ini")
 126         sd2 = config.SwarmDescription()
 127         sd2.load("session_config.ini")
 128
 129         # Check for the same sections and items.
 130         # Use sort() method because self.test_data is unordered.
 131         self.assertEqual(sd2.data.sections().sort(),
 132                          list(self.test_data.keys()).sort())
 133         for section in sd2.data.sections():
 134             self.assertEqual(sd2.data.items(section).sort(),
 135                              list(self.test_data[section]).sort())
 136         os.remove("session_config.ini")
 137
 138     def test_get_swarm(self):
 139         swarm = self.sd.get_swarm()
 140         expected_swarm = storage.Swarm(
 141             self.test_data['swarm']['torrent_filename'],
 142             self.test_data['swarm']['data_size'],
 143             self.test_data['swarm']['description'])
 144
 145         self.assertEqual(swarm.torrent_filename,
 146                          expected_swarm.torrent_filename)
 147         self.assertEqual(swarm.data_size, expected_swarm.data_size)
 148         self.assertEqual(swarm.description, expected_swarm.description)
 149
 150     def test_get_session_entries(self):
 151         session_entries = self.sd.get_session_entries()
 152         expected_session_list = self.test_data.keys()
 153         expected_session_list.remove('swarm')
 154
 155         for entry in session_entries:
 156             section = entry.data.sections()[0]
 157             self.assertEqual(entry.data.items(section).sort(),
 158                              list(self.test_data[section]).sort())
 159
 160     def test_update_session_entry_id(self):
 161         session_entries = self.sd.get_session_entries()
 162         entry = session_entries[0]
 163         cs_id = '1'
 164         self.sd.update_session_entry_id(entry, cs_id)
 165
 166         self.assertEqual(entry.get_session_id(), cs_id)
 167
 168     def test_get_file_archives(self):
 169         archives = self.sd.get_file_archives()
 170         expected_archives = [
 171             'p2p-next-01.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 172             'p2p-next-03.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 173             'p2p-next-04.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 174             'p2p-next-09.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz']
 175
 176         self.assertEqual(archives.sort(),expected_archives.sort())
 177
 178
 179 class SessionEntryTest(unittest.TestCase):
 180
 181     """Test suite for SessionEntry class in config.py."""
 182
 183     # Class specific variables. Initialized in setUp, used throughout tests.
 184     se = None
 185     sd = None
 186     config_file = "config.sample.ini"
 187
 188     def setUp(self):
 189         config_file_path = os.path.join(os.path.dirname(__file__),
 190                                         self.config_file)
 191         self.sd = config.SwarmDescription()
 192         self.sd.load(config_file_path)
 193         session_entries = self.sd.get_session_entries()
 194
 195         self.se = session_entries[2]
 196
 197     def test_get_session(self):
 198         cs = self.se.get_session()
 199
 200         self.assertTrue(type(cs) is storage.ClientSession)
 201         self.assertEqual(cs.system_os, 'Linux')
 202         self.assertEqual(cs.system_ram, '2007')
 203
 204     def test_get_session_id(self):
 205         # Check for no session id before update
 206         #self.assertFalse(self.se.get_session_id())
 207         test_cs_id = 2
 208         self.sd.update_session_entry_id(self.se, test_cs_id)
 209         cs_id = self.se.get_session_id()
 210         self.assertEqual(str(cs_id), str(test_cs_id))
 211
 212
 213 class AccessConfigTest(unittest.TestCase):
 214
 215     """Test suite for AccessConfig class in config.py."""
 216
 217     # Class specific variables. Initialized in setUp, used throughout tests.
 218     access_file = "access.sample.ini"
 219     test_data = None
 220     mysql_create_script = "p2p-log-mysql.sql"
 221     mysql_init_script = "p2p-init-test-mysql.sql"
 222     sqlite_create_script = "p2p-log-sqlite.sql"
 223     sqlite_init_script = "p2p-init-test.sql"
 224     database = {"database": "p2p_test", "user": "root",
 225                 "password": "p2p4th3m45535"}
 226     sqlite_db = "test.db"
 227     ac = None
 228
 229     def setUp(self):
 230         """Instantiate objects and create MySQL and SQLite databases."""
 231         self.test_data = {'mysql': {'user': 'root',
 232                                     'database': 'p2p_test',
 233                                     'password': 'p2p4th3m45535',
 234                                     'host': 'localhost'},
 235                           'sqlite': {'database': 'test.db'}}
 236
 237         # Create MySQL database. Create tables and indices.
 238         sql_create_script_path = os.path.join(os.path.dirname(__file__),
 239                                               self.mysql_create_script)
 240
 241         self.conn = MySQLdb.Connection(user=self.database['user'],
 242                                        passwd=self.database['password'])
 243         self.cursor = self.conn.cursor()
 244
 245         self.cursor.execute("CREATE DATABASE %s" %self.database['database'])
 246         self.cursor.execute("USE %s" %self.database['database'])
 247
 248         f = open(sql_create_script_path, 'r')
 249         p = subprocess.Popen(["mysql", self.database['database'],
 250                               "--user=%s" %self.database['user'],
 251                               "--password=%s" %self.database['password']],
 252                              stdin=f)
 253         ret = os.waitpid(p.pid, 0)[1]
 254         f.close()
 255
 256         # Populate MySQL database.
 257         # TODO: Check exceptions.
 258         sql_init_script_path = os.path.join(os.path.dirname(__file__),
 259                                             self.mysql_init_script)
 260
 261         f = open(sql_init_script_path, 'r')
 262         p = subprocess.Popen(["mysql", self.database['database'],
 263                               "--user=%s" %self.database['user'],
 264                               "--password=%s" %self.database['password']],
 265                              stdin=f)
 266         ret = os.waitpid(p.pid, 0)[1]
 267         f.close()
 268
 269         # Create SQLite database file. Create tables and indices.
 270         # TODO: Check exceptions.
 271         sql_create_script_path = os.path.join(os.path.dirname(__file__),
 272                                               self.sqlite_create_script)
 273         f = open(sql_create_script_path, 'r')
 274         p = subprocess.Popen(["sqlite3", self.sqlite_db], stdin=f)
 275         ret = os.waitpid(p.pid, 0)[1]
 276         f.close()
 277
 278         # Populate database.
 279         # TODO: Check exceptions.
 280         sql_init_script_path = os.path.join(os.path.dirname(__file__),
 281                                             self.sqlite_init_script)
 282         f = open(sql_init_script_path, 'r')
 283         p = subprocess.Popen(["sqlite3", self.sqlite_db], stdin=f)
 284         ret = os.waitpid(p.pid, 0)[1]
 285         f.close()
 286
 287         # Initialize self.ac as AccessConfig instance
 288         access_file_path = os.path.join(os.path.dirname(__file__),
 289                                         self.access_file)
 290         self.ac = config.AccessConfig()
 291         self.ac.load(access_file_path)
 292
 293     def tearDown(self):
 294         """Close connection and remove SQLite / MySQL databases."""
 295         os.remove(self.sqlite_db)
 296
 297         self.cursor.execute("DROP DATABASE %s" %self.database['database'])
 298         self.cursor.close()
 299         self.conn.close()
 300
 301     def test_load(self):
 302         # Check for the same sections.
 303         # Use sort() method because self.test_data is unordered.
 304         self.assertEqual(self.ac.data.sections().sort(),
 305                          list(self.test_data.keys()).sort())
 306         for section in self.ac.data.sections():
 307             self.assertEqual(self.ac.data.items(section).sort(),
 308                              list(self.test_data[section]).sort())
 309
 310     def test_store(self):
 311         self.ac.store("access_config.ini")
 312         ac2 = config.AccessConfig()
 313         ac2.load("access_config.ini")
 314
 315         # Check for the same sections and items.
 316         # Use sort() method because self.test_data is unordered.
 317         self.assertEqual(ac2.data.sections().sort(),
 318                          list(self.test_data.keys()).sort())
 319         for section in ac2.data.sections():
 320             self.assertEqual(ac2.data.items(section).sort(),
 321                              list(self.test_data[section]).sort())
 322         os.remove("access_config.ini")
 323
 324     def test_get_swarm_writer(self):
 325         sw = self.ac.get_swarm_writer()
 326         expected_handlers = 2
 327         handlers = len(sw.handlers)
 328
 329         self.assertEqual(expected_handlers, handlers)
 330
 331
 332 if __name__ == "__main__":
 333     unittest.main()