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

   1 # Written by Boudewijn Schoon, Arno Bakker\r
   2 # see LICENSE.txt for license information\r
   3 \r
   4 import unittest\r
   5 import sys\r
   6 import time\r
   7 from traceback import print_exc\r
   8 from M2Crypto import EC\r
   9 \r
  10 from BaseLib.Test.test_crawler import TestCrawler\r
  11 from olconn import OLConnection\r
  12 from BaseLib.Core.BitTornado.bencode import bencode,bdecode\r
  13 \r
  14 from BaseLib.Core.BitTornado.BT1.MessageID import CRAWLER_FRIENDSHIP_STATS\r
  15 from BaseLib.Core.CacheDB.sqlitecachedb import bin2str\r
  16 from BaseLib.Core.CacheDB.SqliteCacheDBHandler import CrawlerDBHandler\r
  17 from BaseLib.Core.CacheDB.SqliteFriendshipStatsCacheDB import FriendshipStatisticsDBHandler\r
  18 \r
  19 DEBUG=True\r
  20 \r
  21 class TestFriendshipCrawler(TestCrawler):\r
  22     """ \r
  23     Testing the user side of the crawler\r
  24     """\r
  25 \r
  26     def setUpPreSession(self):\r
  27         TestCrawler.setUpPreSession(self)\r
  28         self.config.set_social_networking(True)\r
  29 \r
  30     def setUpPostSession(self):\r
  31         """ override TestAsServer """\r
  32         TestCrawler.setUpPostSession(self)\r
  33 \r
  34         self.some_keypair = EC.gen_params(EC.NID_sect233k1)\r
  35         self.some_keypair.gen_key()\r
  36         self.some_permid = str(self.some_keypair.pub().get_der())\r
  37 \r
  38         self.friendshipStatistics_db = FriendshipStatisticsDBHandler.getInstance()\r
  39         self.friendshipStatistics_db.insertFriendshipStatistics( bin2str(self.his_permid), bin2str(self.some_permid), int(time.time()), 0, commit=True)        \r
  40         self.friendshipStatistics_db.insertFriendshipStatistics( bin2str(self.my_permid), bin2str(self.some_permid), int(time.time()), 0, commit=True)\r
  41 \r
  42         # make sure that the OLConnection IS in the crawler_db\r
  43         crawler_db = CrawlerDBHandler.getInstance()\r
  44         crawler_db.temporarilyAddCrawler(self.my_permid)\r
  45 \r
  46     \r
  47     def test_all(self):\r
  48         """\r
  49         I want to start a Tribler client once and then connect to it\r
  50         many times. So there must be only one test method to prevent\r
  51         setUp() from creating a new client every time.\r
  52 \r
  53         The code is constructed so unittest will show the name of the\r
  54         (sub)test where the error occured in the traceback it prints.\r
  55         """\r
  56         self.subtest_good_friendship_stats()\r
  57 \r
  58     def subtest_good_friendship_stats(self):\r
  59         """\r
  60         Send a valid message-id from a registered crawler peer\r
  61         """\r
  62         print >>sys.stderr, "-"*80, "\ntest: good friendship stats"\r
  63 \r
  64         s = OLConnection(self.my_keypair, "localhost", self.hisport)\r
  65 \r
  66         t = time.time() - 100.0\r
  67         msg_dict = {'current time':int(t)}\r
  68         payload = bencode(msg_dict)\r
  69         self.send_crawler_request(s, CRAWLER_FRIENDSHIP_STATS, 0, 0, payload)\r
  70 \r
  71         error, payload = self.receive_crawler_reply(s, CRAWLER_FRIENDSHIP_STATS, 0)\r
  72         assert error == 0\r
  73         \r
  74         d = bdecode(payload)\r
  75         if DEBUG:\r
  76             print >>sys.stderr, "test: Got FRIENDSHIPSTATISTICS",`d`\r
  77         stats = d['stats']\r
  78         self.assert_(len(stats) == 1)\r
  79         record = d['stats'][0]\r
  80         self.assert_(record[0] == bin2str(self.his_permid))  # source_permid\r
  81         self.assert_(record[1] == bin2str(self.some_permid)) # target_permid\r
  82         self.assert_(record[2] == 0) # isForwarder\r
  83 \r
  84         time.sleep(1)\r
  85         s.close()\r
  86 \r
  87     # def send_crawler_request(self, sock, message_id, channel_id, frequency, payload):\r
  88     # def receive_crawler_reply(self, sock, message_id, channel_id):\r
  89 \r
  90 def test_suite():\r
  91     suite = unittest.TestSuite()\r
  92     suite.addTest(unittest.makeSuite(TestFriendshipCrawler))\r
  93     \r
  94     return suite\r
  95 \r
  96 if __name__ == "__main__":\r
  97     unittest.main(defaultTest="test_suite")\r
  98 \r
  99 \r