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
  13 import config
  14 import storage
  15
  16 class SwarmDescriptionTest(unittest.TestCase):
  17     """
  18     Test suite for SwarmDescription class in config.py.
  19     """
  20
  21     # Class specific variables. Initialized in setUp, used throughout tests.
  22     config_file = "config.sample.ini"
  23     test_data = None
  24     sd = None
  25
  26     def setUp(self):
  27         self.test_data = {'swarm': {'torrent_filename': 'fedora-11-cds.torrent',
  28                                     'data_size': '12345678',
  29                                     'description': 'BitTorrent Distribution Experiment, Fedora 11, i386 netinst'},
  30                           'p2p-next-01': {'bittorrent_client': 'libtorrent-rasterbar',
  31                                           'system_os': 'Linux',
  32                                           'system_os_version': '2.6.26',
  33                                           'system_ram': '1999',
  34                                           'system_cpu': '2992',
  35                                           'public_ip': '141.85.224.201',
  36                                           'public_port': '6881',
  37                                           'ds_limit': '0',
  38                                           'us_limit': '0',
  39                                           'start_time': 'Thu Aug 11 15:36:35 EEST 2010',
  40                                           'dht_enabled': 'true',
  41                                           'pxe_enabled': 'false',
  42                                           'streaming_enabled': 'false',
  43                                           'description': 'NCIT Cluster System',
  44                                           'log_file': 'p2p-next-01.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz'},
  45                           'p2p-next-03': {'bittorrent_client': 'libtorrent-rasterbar',
  46                                           'system_os': 'Linux',
  47                                           'system_os_version': '2.6.26',
  48                                           'system_ram': '2007',
  49                                           'system_cpu': '2992',
  50                                           'public_ip': '141.85.224.209',
  51                                           'public_port': '6881',
  52                                           'ds_limit': '0',
  53                                           'us_limit': '0',
  54                                           'start_time': 'Thu Aug 11 15:36:35 EEST 2010',
  55                                           'dht_enabled': 'true',
  56                                           'pxe_enabled': 'false',
  57                                           'streaming_enabled': 'false',
  58                                           'description': 'NCIT Cluster System',
  59                                           'log_file': 'p2p-next-03.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz'},
  60                           'p2p-next-04': {'bittorrent_client': 'libtorrent-rasterbar',
  61                                           'system_os': 'Linux',
  62                                           'system_os_version': '2.6.26',
  63                                           'system_ram': '2007',
  64                                           'system_cpu': '2992',
  65                                           'public_ip': '141.85.224.204',
  66                                           'public_port': '6881',
  67                                           'ds_limit': '0',
  68                                           'us_limit': '0',
  69                                           'start_time': 'Thu Aug 11 15:36:35 EEST 2010',
  70                                           'dht_enabled': 'true',
  71                                           'pxe_enabled': 'false',
  72                                           'streaming_enabled': 'false',
  73                                           'description': 'NCIT Cluster System',
  74                                           'log_file': 'p2p-next-04.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz'},
  75                          'p2p-next-09': {'bittorrent_client': 'libtorrent-rasterbar',
  76                                           'system_os': 'Linux',
  77                                           'system_os_version': '2.6.26',
  78                                           'system_ram': '1999',
  79                                           'system_cpu': '2992',
  80                                           'public_ip': '141.85.224.209',
  81                                           'public_port': '6881',
  82                                           'ds_limit': '0',
  83                                           'us_limit': '0',
  84                                           'start_time': 'Thu Aug 11 15:36:35 EEST 2010',
  85                                           'dht_enabled': 'true',
  86                                           'pxe_enabled': 'false',
  87                                           'streaming_enabled': 'false',
  88                                           'description': 'NCIT Cluster System',
  89                                           'log_file': 'p2p-next-09.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz'}}
  90
  91         config_file_path = os.path.join(os.path.dirname(__file__),
  92                                         self.config_file)
  93         self.sd = config.SwarmDescription()
  94         self.sd.load(config_file_path)
  95
  96     def test_load(self):
  97         # Check for the same sections and items.
  98         # Use sort() method because self.test_data is unordered.
  99         self.assertEqual(self.sd.data.sections().sort(),
 100                          list(self.test_data.keys()).sort())
 101         for section in self.sd.data.sections():
 102             self.assertEqual(self.sd.data.items(section).sort(),
 103                              list(self.test_data[section]).sort())
 104
 105     def test_store(self):
 106         self.sd.store("session_config.ini")
 107         sd2 = config.SwarmDescription()
 108         sd2.load("session_config.ini")
 109
 110         # Check for the same sections and items.
 111         # Use sort() method because self.test_data is unordered.
 112         self.assertEqual(sd2.data.sections().sort(),
 113                          list(self.test_data.keys()).sort())
 114         for section in sd2.data.sections():
 115             self.assertEqual(sd2.data.items(section).sort(),
 116                              list(self.test_data[section]).sort())
 117         os.remove("session_config.ini")
 118
 119     def test_get_swarm(self):
 120         swarm = self.sd.get_swarm()
 121         expected_swarm = storage.Swarm(
 122                                 self.test_data['swarm']['torrent_filename'],
 123                                 self.test_data['swarm']['data_size'],
 124                                 self.test_data['swarm']['description'])
 125
 126         self.assertEqual(swarm.torrent_filename,
 127                          expected_swarm.torrent_filename)
 128         self.assertEqual(swarm.data_size, expected_swarm.data_size)
 129         self.assertEqual(swarm.description, expected_swarm.description)
 130
 131     def test_get_session_entries(self):
 132         session_entries = self.sd.get_session_entries()
 133         expected_session_list = self.test_data.keys()
 134         expected_session_list.remove('swarm')
 135
 136         for entry in session_entries:
 137             section = entry.data.sections()[0]
 138             self.assertEqual(entry.data.items(section).sort(),
 139                              list(self.test_data[section]).sort())
 140
 141     def test_update_session_entry_id(self):
 142         session_entries = self.sd.get_session_entries()
 143         entry = session_entries[0]
 144         cs_id = '1'
 145         self.sd.update_session_entry_id(entry, cs_id)
 146
 147         self.assertEqual(entry.get_session_id(), cs_id)
 148
 149     def test_get_file_archives(self):
 150         archives = self.sd.get_file_archives()
 151         expected_archives = [
 152                'p2p-next-01.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 153                'p2p-next-03.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 154                'p2p-next-04.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz',
 155                'p2p-next-09.grid.pub.ro/btex-fedora-11/dbex-fedora-cds.tar.gz']
 156
 157         self.assertEqual(archives.sort(),expected_archives.sort())
 158
 159 class SessionEntryTest(unittest.TestCase):
 160     """
 161     Test suite for SessionEntry class in config.py.
 162     """
 163
 164     # Class specific variables. Initialized in setUp, used throughout tests.
 165     se = None
 166     sd = None
 167     config_file = "config.sample.ini"
 168
 169     def setUp(self):
 170         config_file_path = os.path.join(os.path.dirname(__file__),
 171                                         self.config_file)
 172         self.sd = config.SwarmDescription()
 173         self.sd.load(config_file_path)
 174         session_entries = self.sd.get_session_entries()
 175
 176         self.se = session_entries[2]
 177
 178     def test_get_session(self):
 179         cs = self.se.get_session()
 180
 181         self.assertTrue(type(cs) is storage.ClientSession)
 182         self.assertEqual(cs.system_os, 'Linux')
 183         self.assertEqual(cs.system_ram, '2007')
 184
 185     def test_get_session_id(self):
 186         # Check for no session id before update
 187         #self.assertFalse(self.se.get_session_id())
 188         test_cs_id = 2
 189         self.sd.update_session_entry_id(self.se, test_cs_id)
 190         cs_id = self.se.get_session_id()
 191         self.assertEqual(str(cs_id), str(test_cs_id))
 192
 193 class AccessConfigTest(unittest.TestCase):
 194     """
 195     Test suite for AccessConfig class in config.py.
 196     """
 197
 198     # Class specific variables. Initialized in setUp, used throughout tests.
 199     access_file = "access.sample.ini"
 200     test_data = None
 201     ac = None
 202
 203     def setUp(self):
 204         self.test_data = {'mysql': {'user': 'root',
 205                                     'database': 'p2p_test',
 206                                     'password': 'p2p4th3m45535',
 207                                     'host': 'localhost'},
 208                           'sqlite': {'database': 'test.db'}}
 209
 210         access_file_path = os.path.join(os.path.dirname(__file__),
 211                                         self.access_file)
 212         self.ac = config.AccessConfig()
 213         self.ac.load(access_file_path)
 214
 215     def test_load(self):
 216         # Check for the same sections.
 217         # Use sort() method because self.test_data is unordered.
 218         self.assertEqual(self.ac.data.sections().sort(),
 219                          list(self.test_data.keys()).sort())
 220         for section in self.ac.data.sections():
 221             self.assertEqual(self.ac.data.items(section).sort(),
 222                              list(self.test_data[section]).sort())
 223
 224     def test_store(self):
 225         self.ac.store("access_config.ini")
 226         ac2 = config.AccessConfig()
 227         ac2.load("access_config.ini")
 228
 229         # Check for the same sections and items.
 230         # Use sort() method because self.test_data is unordered.
 231         self.assertEqual(ac2.data.sections().sort(),
 232                          list(self.test_data.keys()).sort())
 233         for section in ac2.data.sections():
 234             self.assertEqual(ac2.data.items(section).sort(),
 235                              list(self.test_data[section]).sort())
 236         os.remove("access_config.ini")
 237
 238     def test_get_swarm_writer(self):
 239         sw = self.ac.get_swarm_writer()
 240         expected_handlers = 2
 241         handlers = len(sw.handlers)
 242
 243         self.assertEqual(expected_handlers, handlers)
 244
 245 if __name__ == "__main__":
 246     unittest.main()