instrumentation/next-share/BaseLib/Test/API/test_tracking.py

   1 # Written by Arno Bakker
   2 # see LICENSE.txt for license information
   3 #
   4
   5 import unittest
   6 import os
   7 import sys
   8 import time
   9 import urllib
  10 #from urllib2 import urlopen # urllib blocks on reading, HTTP/1.1 persist conn problem?
  11 from traceback import print_exc
  12 import urlparse
  13
  14 from BaseLib.Core.BitTornado.zurllib import urlopen
  15 from BaseLib.Core.simpledefs import STATEDIR_ITRACKER_DIR
  16 from BaseLib.Test.test_as_server import TestAsServer
  17 from BaseLib.Test.btconn import BTConnection
  18 from BaseLib.Core.BitTornado.BT1.MessageID import *
  19
  20 from BaseLib.Core.API import *
  21
  22 DEBUG=True
  23
  24 class TestTracking(TestAsServer):
  25     """
  26     Testing seeding via new tribler API:
  27     """
  28
  29     def setUpPreSession(self):
  30         """ override TestAsServer """
  31         TestAsServer.setUpPreSession(self)
  32
  33         self.config.set_megacache(False)
  34         self.config.set_internal_tracker(True)
  35
  36
  37     def test_all(self):
  38         self.subtest_add_remove_torrent()
  39         self.subtest_tlookup1()
  40         self.subtest_tlookup2()
  41
  42
  43     def subtest_add_remove_torrent(self):
  44         tdef = TorrentDef()
  45         sourcefn = os.path.join(os.getcwd(),"file.wmv")
  46         tdef.add_content(sourcefn)
  47         tdef.set_tracker(self.session.get_internal_tracker_url())
  48         tdef.finalize()
  49
  50         torrentfn = os.path.join(self.session.get_state_dir(),"gen.torrent")
  51         tdef.save(torrentfn)
  52         infohash = tdef.get_infohash()
  53         hexinfohash = binascii.hexlify(infohash)
  54
  55         self.session.add_to_internal_tracker(tdef)
  56         time.sleep(1)
  57         self.check_http_presence(hexinfohash,True)
  58
  59         self.session.remove_from_internal_tracker(tdef)
  60         print >> sys.stderr,"test: Give network thread running tracker time to detect we removed the torrent file"
  61         time.sleep(2)
  62
  63         self.check_http_presence(hexinfohash,False)
  64         self.check_disk_presence(hexinfohash,False)
  65
  66     def check_http_presence(self,hexinfohash,present):
  67         print >> sys.stderr,"test: infohash is",hexinfohash
  68         url = 'http://127.0.0.1:'+str(self.session.get_listen_port())+'/'
  69         print >> sys.stderr,"test: tracker lives at",url
  70         f = urlopen(url)
  71         data = f.read()
  72         f.close()
  73
  74         # WARNING: this test depends on the output of the tracker. If that
  75         # is changed, also change this.
  76         print >> sys.stderr,"test: tracker returned:",data
  77         if present:
  78             self.assert_(data.find(hexinfohash) != -1)
  79         else:
  80             self.assert_(data.find(hexinfohash) == -1)
  81
  82     def check_disk_presence(self,hexinfohash,present):
  83         itrackerdir = os.path.join(self.session.get_state_dir(),STATEDIR_ITRACKER_DIR)
  84         for filename in os.listdir(itrackerdir):
  85             if filename.startswith(hexinfohash):
  86                 if present:
  87                     self.assert_(True)
  88                 else:
  89                     self.assert_(False)
  90
  91
  92     #
  93     # /tlookup?url
  94     #
  95     def subtest_tlookup1(self):
  96         httpseeds = []
  97         httpseeds.append('http://www.example.com/file.wmv')
  98         self._test_tlookup(httpseeds)
  99
 100     def subtest_tlookup2(self):
 101         httpseeds = []
 102         httpseeds.append('http://www.example.com/file.wmv')
 103         httpseeds.append('http://icecast.freezer.com/file.wmv')
 104         self._test_tlookup(httpseeds)
 105
 106
 107     def _test_tlookup(self,httpseedlist):
 108         t = TorrentDef()
 109         fn = os.path.join(os.getcwd(),"file.wmv")
 110         t.add_content(fn)
 111         t.set_tracker(self.session.get_internal_tracker_url())
 112         t.set_urllist(httpseedlist)
 113         t.finalize()
 114         wantdata = bencode(t.get_metainfo())
 115
 116         self.session.add_to_internal_tracker(t)
 117         #time.sleep(30)
 118
 119         (scheme, netloc, path, pars, query, fragment) = urlparse.urlparse(self.session.get_internal_tracker_url())
 120         urlprefix = scheme+'://'+netloc+'/tlookup?'
 121         for httpseed in httpseedlist:
 122             quoted = urllib.quote(httpseed)
 123             url = urlprefix+quoted
 124             #url = "http://www.cs.vu.nl/~arno/index.html"
 125             print >>sys.stderr,"test: Asking tracker for",url
 126             # F*ing BitTornado/Python crap: using normal urlopen here results in
 127             # an infinitely hanging read (even if read(1024))
 128             conn = urlopen(url)
 129             gotdata = conn.read()
 130             print >>sys.stderr,"test: Tracker sent",len(gotdata)
 131             conn.close()
 132             self.assertEquals(wantdata,gotdata)
 133
 134
 135 def test_suite():
 136     suite = unittest.TestSuite()
 137     suite.addTest(unittest.makeSuite(TestTracking))
 138
 139     return suite
 140
 141 if __name__ == "__main__":
 142     unittest.main()
 143