instrumentation/next-share/BaseLib/Test/test_torrentcollecting.py

   1 import os
   2 import sys
   3 import unittest
   4
   5 from BaseLib.Core.CacheDB.sqlitecachedb import SQLiteCacheDB, str2bin, CURRENT_MAIN_DB_VERSION
   6 from BaseLib.Core.CacheDB.SqliteCacheDBHandler import PreferenceDBHandler, MyPreferenceDBHandler
   7 from BaseLib.Core.BuddyCast.TorrentCollecting import SimpleTorrentCollecting
   8 from bak_tribler_sdb import *
   9
  10 CREATE_SQL_FILE = os.path.join('..',"schema_sdb_v"+str(CURRENT_MAIN_DB_VERSION)+".sql"))
  11 assert os.path.isfile(CREATE_SQL_FILE)
  12
  13 def init():
  14     init_bak_tribler_sdb()
  15
  16
  17 SQLiteCacheDB.DEBUG = False
  18
  19 class TestTorrentCollecting(unittest.TestCase):
  20
  21     def setUp(self):
  22         self.db = SQLiteCacheDB.getInstance()
  23         self.db.initDB(TRIBLER_DB_PATH_BACKUP)
  24
  25         permid = {}
  26         permid[3127] = 'MFIwEAYHKoZIzj0CAQYFK4EEABoDPgAEAcPezgQ13k1MSOaUrCPisWRhYuNT7Tm+q5rUgHFvAWd9b+BcSut6TCniEgHYHDnQ6TH/vxQBqtY8Loag'
  27         permid[994] = 'MFIwEAYHKoZIzj0CAQYFK4EEABoDPgAEAJUNmwvDaigRaM4cj7cE2O7lessqnnFEQsan7df9AZS8xeNmVsP/XXVrEt4t7e2TNicYmjn34st/sx2P'
  28         permid[19] = 'MFIwEAYHKoZIzj0CAQYFK4EEABoDPgAEAAJv2YLuIWa4QEdOEs4CPRxQZDwZphKd/xK/tgbcALG198nNdT10znJ2sZYl+OJIvj7YfYp75PrrnWNX'
  29         permid[5] = 'MFIwEAYHKoZIzj0CAQYFK4EEABoDPgAEAAB0XbUrw5b8CrTrMZST1SPyrzjgSzIE6ynALtlZASGAb+figVXRRGpKW6MSal3KnEm1/q0P3JPWrhCE'
  30         self.permid = permid
  31
  32         db = MyPreferenceDBHandler.getInstance()
  33         db.loadData()
  34
  35     def tearDown(self):
  36         self.db.close()
  37
  38     def test_selecteTorrentToCollect(self):
  39         db = PreferenceDBHandler.getInstance()
  40         tc = SimpleTorrentCollecting(None,None)
  41         truth = {3127:235, 994:20, 19:1, 5:0}
  42
  43         for pid in truth:
  44             pl = db.getPrefList(str2bin(self.permid[pid]))
  45             assert len(pl) == truth[pid], [pid, len(pl)]
  46             # test random selection
  47             infohash = tc.selecteTorrentToCollect(pl, True)
  48             if pid == 994 or pid == 3127:
  49                 assert len(infohash) == 20, infohash
  50             else:
  51                 assert infohash is None, infohash
  52
  53         #tc.updateAllCooccurrence()
  54         for pid in truth:
  55             pl = db.getPrefList(str2bin(self.permid[pid]))
  56             assert len(pl) == truth[pid], [pid, len(pl)]
  57             # test selecting most relevant torrent
  58             infohash = tc.selecteTorrentToCollect(pl, False)
  59             if pid == 994:
  60                 tid = tc.torrent_db.getTorrentID(infohash)
  61                 assert tid == 8979
  62
  63                 permid = self.permid[pid]
  64                 infohash = tc.updatePreferences(permid, pl)
  65                 tid = tc.torrent_db.getTorrentID(infohash)
  66                 assert tid == 8979
  67             elif pid == 3127:
  68                 tid = tc.torrent_db.getTorrentID(infohash)
  69                 assert tid == 9170
  70
  71                 permid = self.permid[pid]
  72                 infohash = tc.updatePreferences(permid, pl)
  73                 tid = tc.torrent_db.getTorrentID(infohash)
  74                 assert tid == 9170
  75             else:
  76                 assert infohash is None, infohash
  77
  78
  79 def test_suite():
  80     suite = unittest.TestSuite()
  81     suite.addTest(unittest.makeSuite(TestTorrentCollecting))
  82
  83     return suite
  84
  85 def main():
  86     init()
  87     unittest.main(defaultTest='test_suite')
  88
  89
  90 if __name__ == '__main__':
  91     main()
  92
  93