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

   1 # Written by Arno Bakker
   2 # see LICENSE.txt for license information
   3 #
   4 # TODO:
   5 #
   6
   7 import sys
   8 import unittest
   9 import os
  10 import tempfile
  11
  12 from BaseLib.Core.API import TorrentDef
  13 from BaseLib.Core.BitTornado.bencode import bdecode
  14 from BaseLib.Core.Utilities.utilities import isValidTorrentFile
  15 from BaseLib.Core.Overlay.permid import verify_torrent_signature
  16
  17 DEBUG = False
  18
  19 TRACKER = 'http://www.tribler.org/announce'
  20 PLAYTIME = "1:02" # 1 min 2 secs. Coordinate with size of file.wmv to get > 1 bitrate
  21 PLAYTIME_SECS = 62 # PLAYTIME in seconds
  22
  23 class TestTorrentDef(unittest.TestCase):
  24     """
  25     Testing TorrentDef version 0
  26     """
  27
  28     def setUp(self):
  29         pass
  30
  31     def tearDown(self):
  32         pass
  33
  34     def test_add_content_file(self):
  35         self.subtest_add_content_file(merkle=False)
  36         self.subtest_add_content_file(merkle=True)
  37
  38     def test_add_content_dir(self):
  39         self.subtest_add_content_dir(merkle=False)
  40         self.subtest_add_content_dir(merkle=True)
  41
  42     def test_add_content_dir_and_file(self):
  43         self.subtest_add_content_dir_and_file(merkle=False)
  44         self.subtest_add_content_dir_and_file(merkle=True)
  45
  46     def test_add_content_file_playtime(self):
  47         self.subtest_add_content_file_playtime(merkle=False)
  48         self.subtest_add_content_file_playtime(merkle=True)
  49
  50     def test_add_content_dir_playtime(self):
  51         self.subtest_add_content_dir_playtime(merkle=False)
  52         self.subtest_add_content_dir_playtime(merkle=True)
  53
  54     def test_add_content_file_thumbnail(self):
  55         self.subtest_add_content_file_thumbnail(merkle=False)
  56         self.subtest_add_content_file_thumbnail(merkle=True)
  57
  58     def test_add_content_announce_list(self):
  59         self.subtest_add_content_announce_list(merkle=False)
  60         self.subtest_add_content_announce_list(merkle=True)
  61
  62     def test_add_content_httpseeds(self):
  63         self.subtest_add_content_httpseeds(merkle=False)
  64         self.subtest_add_content_httpseeds(merkle=True)
  65
  66     def test_add_content_torrentsig(self):
  67         self.subtest_add_content_torrentsig(merkle=False)
  68         self.subtest_add_content_torrentsig(merkle=True)
  69
  70     def test_add_content_piece_length(self):
  71         self.subtest_add_content_piece_length(merkle=False)
  72         self.subtest_add_content_piece_length(merkle=True)
  73
  74     def test_add_content_file_save(self):
  75         self.subtest_add_content_file_save(merkle=False)
  76         self.subtest_add_content_file_save(merkle=True)
  77
  78     def test_ns_metadata(self):
  79         dummydata = "HalloWereld"
  80         t = TorrentDef()
  81         t.set_metadata(dummydata)
  82         fn = os.path.join(os.getcwd(),"file.wmv")
  83         t.add_content(fn)
  84         t.set_tracker(TRACKER)
  85         t.finalize()
  86
  87         [handle,filename] = tempfile.mkstemp()
  88         os.close(handle)
  89         t.save(filename)
  90
  91         t2 = TorrentDef.load(filename)
  92         self.assert_(t2.get_metadata() == dummydata)
  93
  94
  95     def subtest_add_content_file(self,merkle=True):
  96         """ Add a single file to a TorrentDef """
  97         t = TorrentDef()
  98         t.set_create_merkle_torrent(merkle)
  99         fn = os.path.join(os.getcwd(),"file.wmv")
 100         t.add_content(fn)
 101         t.set_tracker(TRACKER)
 102         t.finalize()
 103
 104         s = os.path.getsize("file.wmv")
 105
 106         metainfo = t.get_metainfo()
 107         self.general_check(metainfo)
 108
 109         self.assert_(metainfo['info']['name'] == "file.wmv")
 110         self.assert_(metainfo['info']['length'] == s)
 111
 112         """
 113         bdata = bencode(t.get_metainfo())
 114         f = open("gen.torrent","wb")
 115         f.write(bdata)
 116         f.close()
 117         """
 118
 119     def subtest_add_content_dir(self,merkle=True):
 120         """ Add a single dir to a TorrentDef """
 121         t = TorrentDef()
 122         t.set_create_merkle_torrent(merkle)
 123         dn = os.path.join(os.getcwd(),"contentdir")
 124         t.add_content(dn,"dirintorrent")
 125         t.set_tracker(TRACKER)
 126         t.finalize()
 127
 128         exps = 0L
 129         for f in os.listdir("contentdir"):
 130             if f.startswith('.'):
 131                 continue
 132             p = os.path.join("contentdir",f)
 133             s = os.path.getsize(p)
 134             exps += s
 135             print "test: expected size",f,s
 136
 137         print "test: expected total size of files in torrent",exps
 138
 139         metainfo = t.get_metainfo()
 140         self.general_check(metainfo)
 141
 142         self.assert_(metainfo['info']['name'] == 'dirintorrent')
 143         reals = 0L
 144         for file in metainfo['info']['files']:
 145             s = file['length']
 146             print "test: real size",file['path'],s
 147             reals += s
 148
 149         print "test: real size of files in torrent",reals
 150
 151         self.assert_(exps == reals)
 152
 153
 154     def subtest_add_content_dir_and_file(self,merkle=True):
 155         """ Add a single dir and single file to a TorrentDef """
 156         t = TorrentDef()
 157         t.set_create_merkle_torrent(merkle)
 158
 159         dn = os.path.join(os.getcwd(),"contentdir")
 160         t.add_content(dn,"dirintorrent")
 161
 162         fn = os.path.join(os.getcwd(),"file.wmv")
 163         t.add_content(fn,os.path.join("dirintorrent","file.wmv"))
 164
 165         t.set_tracker(TRACKER)
 166         t.finalize()
 167
 168         # Check
 169         exps = os.path.getsize(fn)
 170         for f in os.listdir("contentdir"):
 171             if f.startswith('.'):
 172                 continue
 173             p = os.path.join("contentdir",f)
 174             exps += os.path.getsize(p)
 175
 176         metainfo = t.get_metainfo()
 177         self.general_check(metainfo)
 178         self.assert_(metainfo['info']['name'] == 'dirintorrent')
 179
 180         reals = 0L
 181         for file in metainfo['info']['files']:
 182             reals += file['length']
 183         self.assert_(exps == reals)
 184
 185
 186     def subtest_add_content_file_playtime(self,merkle=True):
 187         """ Add a single file with playtime to a TorrentDef """
 188         t = TorrentDef()
 189         t.set_create_merkle_torrent(merkle)
 190         fn = os.path.join(os.getcwd(),"file.wmv")
 191         t.add_content(fn,playtime=PLAYTIME)
 192         t.set_tracker(TRACKER)
 193         t.finalize()
 194
 195         s = os.path.getsize("file.wmv")
 196
 197         metainfo = t.get_metainfo()
 198         self.general_check(metainfo)
 199         self.assert_(metainfo['info']['playtime']==PLAYTIME)
 200         azprop = metainfo['azureus_properties']
 201         content = azprop['Content']
 202         realspeedbps = content['Speed Bps']
 203         expspeedbps = s/PLAYTIME_SECS
 204         self.assert_(realspeedbps == expspeedbps)
 205
 206
 207     def subtest_add_content_dir_playtime(self,merkle=True):
 208         """ Add a single dir to a TorrentDef """
 209         t = TorrentDef()
 210         t.set_create_merkle_torrent(merkle)
 211         fn1 = os.path.join(os.getcwd(),"contentdir","file.avi")
 212         fn2 = os.path.join(os.getcwd(),"contentdir","file.txt")
 213         t.add_content(fn1,os.path.join("dirintorrent","file.avi"),playtime=PLAYTIME)
 214         t.add_content(fn2,os.path.join("dirintorrent","file.txt"))
 215         t.set_tracker(TRACKER)
 216         t.finalize()
 217
 218         metainfo = t.get_metainfo()
 219         self.general_check(metainfo)
 220         self.assert_(metainfo['info']['name'] == 'dirintorrent')
 221
 222         s = os.path.getsize(fn1)
 223
 224         files = metainfo['info']['files']
 225         for file in files:
 226             if file['path'][0] == "file.avi":
 227                 self.assert_(file['playtime']==PLAYTIME)
 228
 229         azprop = metainfo['azureus_properties']
 230         content = azprop['Content']
 231         realspeedbps = content['Speed Bps']
 232         expspeedbps = s/PLAYTIME_SECS
 233         self.assert_(realspeedbps == expspeedbps)
 234
 235
 236     def subtest_add_content_file_thumbnail(self,merkle=True):
 237         """ Add a single file with thumbnail to a TorrentDef """
 238         t = TorrentDef()
 239         t.set_create_merkle_torrent(merkle)
 240         fn = os.path.join(os.getcwd(),"file.wmv")
 241         thumbfn = os.path.join(os.getcwd(),"thumb.jpg")
 242         t.add_content(fn)
 243         t.set_thumbnail(thumbfn)
 244         t.set_tracker(TRACKER)
 245         t.finalize()
 246
 247         f = open(thumbfn,"rb")
 248         expthumb = f.read()
 249         f.close()
 250
 251         metainfo = t.get_metainfo()
 252         self.general_check(metainfo)
 253         azprop = metainfo['azureus_properties']
 254         content = azprop['Content']
 255         realthumb = content['Thumbnail']
 256         self.assert_(realthumb == expthumb)
 257
 258
 259     def subtest_add_content_announce_list(self,merkle=True):
 260         """ Add a single file with announce-list to a TorrentDef """
 261         t = TorrentDef()
 262         t.set_create_merkle_torrent(merkle)
 263         fn = os.path.join(os.getcwd(),"file.wmv")
 264         t.add_content(fn)
 265         t.set_tracker(TRACKER)
 266         exphier = [[TRACKER],['http://tracker1.tribler.org:6969/announce','http://tracker2.tribler.org:7070/ann'],['http://www.cs.vu.nl','http://www.st.ewi.tudelft.nl','http://www.vuze.com']]
 267         t.set_tracker_hierarchy(exphier)
 268         t.finalize()
 269
 270         metainfo = t.get_metainfo()
 271         self.general_check(metainfo)
 272         realhier = metainfo['announce-list']
 273         self.assert_(realhier == exphier)
 274
 275
 276     def subtest_add_content_httpseeds(self,merkle=True):
 277         """ Add a single file with BitTornado httpseeds to a TorrentDef """
 278         t = TorrentDef()
 279         t.set_create_merkle_torrent(merkle)
 280         fn = os.path.join(os.getcwd(),"file.wmv")
 281         t.add_content(fn)
 282         t.set_tracker(TRACKER)
 283         expseeds = ['http://www.cs.vu.nl/index.html','http://www.st.ewi.tudelft.nl/index.html']
 284         t.set_httpseeds(expseeds)
 285         t.finalize()
 286
 287         metainfo = t.get_metainfo()
 288         self.general_check(metainfo)
 289         realseeds = metainfo['httpseeds']
 290         self.assert_(realseeds == expseeds)
 291
 292     def subtest_add_content_torrentsig(self,merkle=True):
 293         """ Add a single file to a TorrentDef and sign it """
 294         t = TorrentDef()
 295         t.set_create_merkle_torrent(merkle)
 296         fn = os.path.join(os.getcwd(),"file.wmv")
 297         kpfn = os.path.join(os.getcwd(),"ec.pem")
 298         t.add_content(fn)
 299         t.set_signature_keypair_filename(kpfn)
 300         t.set_tracker(TRACKER)
 301         t.finalize()
 302
 303         metainfo = t.get_metainfo()
 304         self.general_check(metainfo)
 305         ret = verify_torrent_signature(metainfo)
 306         self.assert_(ret == True)
 307
 308
 309     def subtest_add_content_piece_length(self,merkle=True):
 310         """ Add a single file with piece length to a TorrentDef """
 311         t = TorrentDef()
 312         t.set_create_merkle_torrent(merkle)
 313         fn = os.path.join(os.getcwd(),"file.wmv")
 314         t.add_content(fn)
 315         t.set_piece_length(2 ** 16)
 316         t.set_tracker(TRACKER)
 317         t.finalize()
 318
 319         metainfo = t.get_metainfo()
 320         self.general_check(metainfo)
 321         self.assert_(metainfo['info']['piece length'] == 2 ** 16)
 322
 323
 324     def subtest_add_content_file_save(self,merkle=True):
 325         """ Add a single file to a TorrentDef and save the latter"""
 326         t = TorrentDef()
 327         t.set_create_merkle_torrent(merkle)
 328         fn = os.path.join(os.getcwd(),"file.wmv")
 329         t.add_content(fn)
 330         t.set_tracker(TRACKER)
 331         t.finalize()
 332
 333         tfn = os.path.join(os.getcwd(),"gen.torrent")
 334         t.save(tfn)
 335
 336         f = open(tfn,"rb")
 337         bdata = f.read()
 338         f.close()
 339         os.remove(tfn)
 340
 341         data = bdecode(bdata)
 342         metainfo = t.get_metainfo()
 343         self.general_check(metainfo)
 344         self.assert_(metainfo == data)
 345
 346
 347     def general_check(self,metainfo):
 348         self.assert_(isValidTorrentFile(metainfo))
 349         self.assert_(metainfo['announce'] == TRACKER)
 350
 351
 352
 353
 354 def test_suite():
 355     suite = unittest.TestSuite()
 356     suite.addTest(unittest.makeSuite(TestTorrentDef))
 357
 358     return suite
 359
 360 if __name__ == "__main__":
 361     unittest.main()
 362