instrumentation/next-share/BaseLib/Test/API/test_seeding_vod.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 socket
  10 import tempfile
  11
  12 from BaseLib.Test.test_as_server import TestAsServer
  13 from BaseLib.Test.btconn import BTConnection
  14 from BaseLib.Core.BitTornado.BT1.MessageID import *
  15
  16 from BaseLib.Core.TorrentDef import *
  17 from BaseLib.Core.DownloadConfig import *
  18 from BaseLib.Core.Session import *
  19 from BaseLib.Core.simpledefs import *
  20
  21 DEBUG=True
  22
  23 class TestSeeding(TestAsServer):
  24     """
  25     Testing seeding via new tribler API:
  26     """
  27
  28     def setUp(self):
  29         """ override TestAsServer """
  30         TestAsServer.setUp(self)
  31         print >>sys.stderr,"test: Giving Session time to startup"
  32         time.sleep(5)
  33         print >>sys.stderr,"test: Session should have started up"
  34         self.vod_started = False
  35
  36     def setUpPreSession(self):
  37         """ override TestAsServer """
  38         TestAsServer.setUpPreSession(self)
  39
  40         self.config.set_overlay(False)
  41         self.config.set_internal_tracker(True)
  42
  43         self.mylistenport = 4810
  44
  45     def setUpPostSession(self):
  46         pass
  47
  48     def tearDown(self):
  49         TestAsServer.tearDown(self)
  50         self.assert_(self.vod_started)
  51
  52     def setup_seeder(self,merkle):
  53         self.tdef = TorrentDef()
  54         self.sourcefn = os.path.join(os.getcwd(),"file2.wmv")
  55         self.tdef.add_content(self.sourcefn,playtime='1:00') # 60 secs
  56         self.tdef.set_create_merkle_torrent(merkle)
  57         self.tdef.set_tracker(self.session.get_internal_tracker_url())
  58         self.tdef.finalize()
  59
  60         self.torrentfn = os.path.join(self.session.get_state_dir(),"gen.torrent")
  61         self.tdef.save(self.torrentfn)
  62
  63         print >>sys.stderr,"test: setup_seeder: name is",self.tdef.metainfo['info']['name']
  64
  65         self.dscfg = DownloadStartupConfig()
  66         self.dscfg.set_dest_dir(os.getcwd())
  67         d = self.session.start_download(self.tdef,self.dscfg)
  68
  69         d.set_state_callback(self.seeder_state_callback)
  70
  71     def seeder_state_callback(self,ds):
  72         d = ds.get_download()
  73         print >>sys.stderr,"test: seeder:",`d.get_def().get_name()`,dlstatus_strings[ds.get_status()],ds.get_progress()
  74         return (1.0,False)
  75
  76
  77     def test_normal_torrent(self):
  78         """
  79             I want to start a Tribler client once and then connect to
  80             it many times. So there must be only one test method
  81             to prevent setUp() from creating a new client every time.
  82
  83             The code is constructed so unittest will show the name of the
  84             (sub)test where the error occured in the traceback it prints.
  85         """
  86         self.setup_seeder(False)
  87         #self.subtest_is_seeding()
  88         self.subtest_download()
  89
  90     def test_merkle_torrent(self):
  91         self.setup_seeder(True)
  92         self.subtest_is_seeding()
  93         self.subtest_download()
  94
  95     def subtest_is_seeding(self):
  96         infohash = self.tdef.get_infohash()
  97         s = BTConnection('localhost',self.hisport,user_infohash=infohash)
  98         s.read_handshake_medium_rare()
  99
 100         s.send(CHOKE)
 101         try:
 102             s.s.settimeout(10.0)
 103             resp = s.recv()
 104             self.assert_(len(resp) > 0)
 105             self.assert_(resp[0] == EXTEND)
 106         except socket.timeout:
 107             print >> sys.stderr,"test: Timeout, peer didn't reply"
 108             self.assert_(False)
 109         s.close()
 110
 111
 112     def subtest_download(self):
 113         """ Now download the file via another Session """
 114
 115         self.config2 = self.config.copy() # not really necess
 116         self.config_path2 = tempfile.mkdtemp()
 117         self.config2.set_state_dir(self.config_path2)
 118         self.config2.set_listen_port(self.mylistenport)
 119         self.session2 = Session(self.config2,ignore_singleton=True)
 120
 121         # Allow session2 to start
 122         print >>sys.stderr,"test: Sleeping 3 secs to let Session2 start"
 123         time.sleep(3)
 124
 125         tdef2 = TorrentDef.load(self.torrentfn)
 126
 127         dscfg2 = DownloadStartupConfig()
 128         dscfg2.set_dest_dir(self.config_path2)
 129         dscfg2.set_video_event_callback(self.downloader_vod_ready_callback)
 130
 131         d = self.session2.start_download(tdef2,dscfg2)
 132         d.set_state_callback(self.downloader_state_callback)
 133         time.sleep(20)
 134
 135     def downloader_state_callback(self,ds):
 136         d = ds.get_download()
 137         print >>sys.stderr,"test: download:",`d.get_def().get_name()`,dlstatus_strings[ds.get_status()],ds.get_progress()
 138
 139         return (1.0,False)
 140
 141     def downloader_vod_ready_callback(self,d,event,params):
 142         if event == VODEVENT_START:
 143             self.vod_started = True
 144
 145
 146 def test_suite():
 147     suite = unittest.TestSuite()
 148     # We should run the tests in a separate Python interpreter to prevent
 149     # problems with our singleton classes, e.g. PeerDB, etc.
 150     if len(sys.argv) != 2:
 151         print "Usage: python test_seeding.py <method name>"
 152     else:
 153         suite.addTest(TestSeeding(sys.argv[1]))
 154
 155     return suite
 156
 157 def main():
 158     unittest.main(defaultTest='test_suite',argv=[sys.argv[0]])
 159
 160 if __name__ == "__main__":
 161     main()