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

   1 # Written by Andrea Reale
   2 # see LICENSE.txt for license information
   3
   4 from __future__ import with_statement
   5 from BaseLib.Test.test_as_server import TestAsServer
   6 import tempfile
   7 from Tribler import LIBRARYNAME
   8 import os
   9 import sys
  10 import shutil
  11 import hashlib
  12 from BaseLib.Core.BitTornado.BT1.MessageID import GET_SUBS, SUBS
  13 from BaseLib.Core.BitTornado.bencode import bencode, bdecode
  14 from BaseLib.Core.Overlay.permid import generate_keypair
  15 from BaseLib.Core.simpledefs import NTFY_RICH_METADATA
  16 from BaseLib.Core.Subtitles.MetadataDomainObjects.MetadataDTO import MetadataDTO
  17 from BaseLib.Core.Subtitles.MetadataDomainObjects.SubtitleInfo import SubtitleInfo
  18 from traceback import print_exc
  19 from BaseLib.Core.Subtitles.MetadataDomainObjects.Languages import LanguagesProvider
  20 from BaseLib.Core.Utilities import utilities
  21 from BaseLib.Core.Subtitles.SubtitlesHandler import SubtitlesHandler
  22 import BaseLib.Core.Subtitles.SubtitlesHandler as SubUtils
  23 from BaseLib.Test.olconn import OLConnection
  24 from BaseLib.Core.BuddyCast.buddycast import BuddyCastCore
  25 from BaseLib.Core.BuddyCast.channelcast import ChannelCastCore
  26 from BaseLib.Core.BuddyCast.votecast import VoteCastCore
  27 import codecs
  28 import unittest
  29
  30 DEBUG = False
  31 RES_DIR = os.path.join('.', 'subtitles_test_res')
  32
  33 class TestSubtitleMessages(TestAsServer):
  34
  35     def setUpPreSession(self):
  36         """ override TestAsServer """
  37         TestAsServer.setUpPreSession(self)
  38         self.config.set_buddycast(True)
  39         BuddyCastCore.TESTASSERVER = True
  40         ChannelCastCore.TESTASSERVER = True
  41         VoteCastCore.TESTASSERVER = True
  42         self.config.set_start_recommender(True)
  43         self.config.set_bartercast(True)
  44         self.config.set_remote_query(True)
  45         self.config.set_crawler(False)
  46         self.config.set_torrent_collecting_dir(os.path.join(self.config_path, "tmp_torrent_collecting"))
  47
  48         self.collecting_dir = os.path.join(self.config_path, "temp_subtitles_collecting")
  49         os.makedirs(self.collecting_dir)
  50         self.config.set_subtitles_collecting(True)
  51         self.config.set_subtitles_collecting_dir(self.collecting_dir)
  52
  53
  54
  55 #        # Write superpeers.txt and DB schema
  56         self.install_path = tempfile.mkdtemp()
  57         spdir = os.path.join(self.install_path, LIBRARYNAME, 'Core')
  58         os.makedirs(spdir)
  59
  60         statsdir = os.path.join(self.install_path, LIBRARYNAME, 'Core', 'Statistics')
  61         os.makedirs(statsdir)
  62
  63         superpeerfilename = os.path.join(spdir, 'superpeer.txt')
  64         print >> sys.stderr,"test: writing empty superpeers to",superpeerfilename
  65         f = open(superpeerfilename, "w")
  66         f.write('# Leeg')
  67         f.close()
  68
  69         self.config.set_install_dir(self.install_path)
  70
  71         srcfiles = []
  72         srcfiles.append(os.path.join(LIBRARYNAME,"schema_sdb_v5.sql"))
  73         for srcfile in srcfiles:
  74             sfn = os.path.join('..','..',srcfile)
  75             dfn = os.path.join(self.install_path,srcfile)
  76             print >>sys.stderr,"test: copying",sfn,dfn
  77             shutil.copyfile(sfn,dfn)
  78
  79         #copy subtitles files in the appropriate subtitles folder
  80         self.src1 = os.path.join(RES_DIR,'fake.srt')
  81         self.src2 = os.path.join(RES_DIR,'fake0.srt')
  82
  83
  84
  85
  86     def setUpPostSession(self):
  87         """ override TestAsServer """
  88         TestAsServer.setUpPostSession(self)
  89
  90         self.mypermid = str(self.my_keypair.pub().get_der())
  91         self.hispermid = str(self.his_keypair.pub().get_der())
  92
  93         self.another_keypair = generate_keypair()
  94         self.anotherpermid = str(self.another_keypair.pub().get_der())
  95
  96         self.testInfohash = hashlib.sha1("yoman!").digest()
  97
  98         #copy subtitles in the collecting dir
  99         nldName = SubUtils.getSubtitleFileRelativeName(self.anotherpermid, self.testInfohash, "nld")
 100         engName = SubUtils.getSubtitleFileRelativeName(self.anotherpermid, self.testInfohash, "eng")
 101
 102         self.sub1 = os.path.join(self.collecting_dir, nldName)
 103         self.sub2 = os.path.join(self.collecting_dir, engName)
 104
 105         shutil.copyfile(self.src1, self.sub1)
 106         # Let's say that the receiving peer has only the nld subtitle
 107         # avialable locally
 108         # shutil.copyfile(self.src2, self.sub2)
 109
 110
 111     def setUpDB(self):
 112         try:
 113             self.richMetadata_db = self.session.open_dbhandler(NTFY_RICH_METADATA)
 114
 115             #add some metadata
 116             self.mdto = MetadataDTO(self.anotherpermid, self.testInfohash)
 117             subtitle1 = SubtitleInfo("nld", self.sub1)
 118             subtitle1.computeChecksum()
 119             subtitle2 = SubtitleInfo("eng", os.path.join(RES_DIR, "fake0.srt"))
 120             subtitle2.computeChecksum()
 121             self.mdto.addSubtitle(subtitle1)
 122             self.mdto.addSubtitle(subtitle2)
 123
 124             self.mdto.sign(self.another_keypair)
 125
 126             self.richMetadata_db.insertMetadata(self.mdto)
 127
 128
 129
 130             #hisoermid has the nld subtitle but doesn't have the english one
 131             self.richMetadata_db.updateSubtitlePath(self.mdto.channel,self.mdto.infohash,"eng",None)
 132
 133         except:
 134             print_exc()
 135
 136
 137     def tearDown(self):
 138         TestAsServer.tearDown(self)
 139         self.session.close_dbhandler(self.richMetadata_db)
 140
 141
 142
 143     def subtest_receptionOfSUBS(self):
 144         '''
 145         Asking for the single available subtitle. The response should be
 146         a valid SUBS message containing its contents
 147         '''
 148
 149         print >> sys.stderr, "test: test_subtitles_msgs_1_1 -----------------------"
 150         ol_conn = OLConnection(self.my_keypair,'localhost',self.hisport)
 151
 152         bitmask = LanguagesProvider.getLanguagesInstance().langCodesToMask(['nld'])
 153         binmask = utilities.uintToBinaryString(bitmask, length=4)
 154
 155         request = GET_SUBS + \
 156                       bencode((
 157                               self.anotherpermid,
 158                               self.testInfohash,
 159                               binmask
 160                               ))
 161
 162         subshandler = SubtitlesHandler()
 163         subshandler.register(ol_conn, self.richMetadata_db, self.session)
 164
 165         ol_conn.send(request)
 166         subs_data = ol_conn.recv()
 167         print >> sys.stderr, "test: subtitles_messages : received SUBS response: len",len(subs_data)
 168         self.assertEquals(SUBS, subs_data[0])
 169         data = bdecode(subs_data[1:])
 170         print >> sys.stderr, "test: subtitles_messages : received SUBS response: ", data
 171
 172         #check on the format of the response
 173         self.assertTrue(isinstance(data,list))
 174         self.assertEquals(4, len(data)) # for fields
 175         self.assertEquals(self.mdto.channel,data[0])
 176         self.assertEquals(self.mdto.infohash, data[1])
 177         self.assertEquals(binmask, data[2])
 178         self.assertTrue(isinstance(data[3],list))
 179         self.assertEquals(1, len(data[3]))
 180         with codecs.open(self.sub1, "rb", "utf-8") as sub:
 181             expectedContents = sub.read()
 182         self.assertEquals(expectedContents, data[3][0])
 183
 184         ol_conn.close()
 185
 186         print >> sys.stderr, "test: subtitles_messages: received content is valid."
 187         print >> sys.stderr, "End of test_subtitles_msgs_1_1 test --------------------"
 188
 189
 190     def subtest_receptionOfSUBSTwoRequestsOneAvailable(self):
 191         """
 192         Asking for two subtitles while the recipent of the request has only one.
 193         The response should contain only the one available subtitle content,
 194         plus a bitmask that reflects the contents of the response.
 195         """
 196
 197         print >> sys.stderr, "test: test_subtitles_msgs_2_1 -----------------------"
 198         ol_conn = OLConnection(self.my_keypair,'localhost',self.hisport)
 199
 200         bitmask = LanguagesProvider.getLanguagesInstance().langCodesToMask(['nld','eng'])
 201         binmask = utilities.uintToBinaryString(bitmask, length=4)
 202
 203         request = GET_SUBS + \
 204                       bencode((
 205                               self.anotherpermid,
 206                               self.testInfohash,
 207                               binmask
 208                               ))
 209
 210         subshandler = SubtitlesHandler()
 211         subshandler.register(ol_conn, self.richMetadata_db, self.session)
 212
 213         ol_conn.send(request)
 214         subs_data = ol_conn.recv()
 215         self.assertEquals(SUBS, subs_data[0])
 216         data = bdecode(subs_data[1:])
 217         print >> sys.stderr, "test: subtitles_messages : received SUBS repsonse: ", data
 218
 219         #check on the format of the response
 220         self.assertTrue(isinstance(data,list))
 221         self.assertEquals(4, len(data)) # for fields
 222         self.assertEquals(self.mdto.channel,data[0])
 223         self.assertEquals(self.mdto.infohash, data[1])
 224
 225         #the receiver had only one of the two requested subtitles
 226         # so I expect a different bitmask
 227         bitmask = LanguagesProvider.getLanguagesInstance().langCodesToMask(['nld'])
 228         expectedBinarymask = utilities.uintToBinaryString(bitmask, length=4)
 229
 230         self.assertEquals(expectedBinarymask, data[2])
 231         self.assertTrue(isinstance(data[3],list))
 232         self.assertEquals(1, len(data[3]))
 233         with codecs.open(self.sub1, "rb", "utf-8") as sub:
 234             expectedContents = sub.read()
 235         self.assertEquals(expectedContents, data[3][0])
 236
 237         ol_conn.close()
 238         print >> sys.stderr, "test: subtitles_messages: received content is valid."
 239         print >> sys.stderr, "End of test_subtitles_msgs_2_1 test --------------------"
 240
 241     def subtest_invalidRequest1(self):
 242         """
 243         Trying to send an empty message.
 244         The connection should be closed by the receiver
 245         """
 246         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_1 ------------------"
 247         ol_conn = OLConnection(self.my_keypair,'localhost',self.hisport)
 248
 249
 250         request = GET_SUBS + \
 251                     bencode({})
 252
 253         ol_conn.send(request)
 254         self.assertEquals(0, len(ol_conn.recv()))
 255         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_1: connection closed as expected"
 256
 257         ol_conn.close()
 258         print >> sys.stderr, "End of test_subtitles_msgs_invalid_request_1 ------------------"
 259
 260     def subtest_invalidRequest2(self):
 261         """
 262         Trying to send an invalid message (an integer instead of a 4 bytes binary string)
 263         The connection should be closed by the receiver
 264         """
 265         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_2 ------------------"
 266         ol_conn = OLConnection(self.my_keypair,'localhost',self.hisport)
 267
 268
 269         request = GET_SUBS + \
 270                       bencode((
 271                               self.anotherpermid,
 272                               self.testInfohash,
 273                               42
 274                               ))
 275
 276         ol_conn.send(request)
 277         self.assertEquals(0, len(ol_conn.recv()))
 278         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_2: connection closed as expected"
 279
 280         ol_conn.close()
 281         print >> sys.stderr, "End of test_subtitles_msgs_invalid_request_2 ------------------"
 282
 283     def subtest_invalidRequest3(self):
 284         """
 285         Trying to send an invalid message (valid for everythin except that there is one field more)
 286         The connection should be closed by the receiver
 287         """
 288         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_3 ------------------"
 289         ol_conn = OLConnection(self.my_keypair,'localhost',self.hisport)
 290
 291         bitmask = LanguagesProvider.getLanguagesInstance().langCodesToMask(['nld','eng'])
 292         binmask = utilities.uintToBinaryString(bitmask, length=4)
 293
 294         request = GET_SUBS + \
 295                       bencode((
 296                               self.anotherpermid,
 297                               self.testInfohash,
 298                               binmask,
 299                               42
 300                               ))
 301
 302         ol_conn.send(request)
 303         self.assertEquals(0, len(ol_conn.recv()))
 304         print >> sys.stderr, "test: test_subtitles_msgs_invalid_request_3: connection closed as expected"
 305
 306         ol_conn.close()
 307         print >> sys.stderr, "End of test_subtitles_msgs_invalid_request_3 ------------------"
 308
 309     def singtest_subs_messages(self):
 310         self.setUpDB()
 311
 312         self.subtest_receptionOfSUBS()
 313         self.subtest_receptionOfSUBSTwoRequestsOneAvailable()
 314         self.subtest_invalidRequest1()
 315         self.subtest_invalidRequest2()
 316         self.subtest_invalidRequest3()
 317
 318         #testMethods = [getattr(self, method) for method in dir(self) if method.startswith('subtest')]
 319
 320         #for m in testMethods:
 321          #   m()
 322
 323
 324 def test_suite():
 325     suite = unittest.TestSuite()
 326     # We should run the tests in a separate Python interpreter to prevent
 327     # problems with our singleton classes, e.g. PeerDB, etc.
 328     if len(sys.argv) != 2:
 329         print "Usage: python test_channelcast_plus_subtitles.py <method name>"
 330     else:
 331         suite.addTest(TestSubtitleMessages(sys.argv[1]))
 332
 333     return suite
 334
 335 def main():
 336     unittest.main(defaultTest='test_suite',argv=[sys.argv[0]])
 337
 338 if __name__ == "__main__":
 339     main()
 340
 341
 342
 343