<s> """<STR_LIT>""" <EOL> import StringIO <EOL> try : <EOL> from twisted . web import client <EOL> except ImportError : <EOL> msn = None <EOL> else : <EOL> from twisted . words . protocols import msn <EOL> from twisted . python . hashlib import md5 <EOL> from twisted . protocols import loopback <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . trial import unittest <EOL> from twisted . test . proto_helpers import StringTransport , StringIOWithoutClosing <EOL> def printError ( f ) : <EOL> print f <EOL> class PassportTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . result = [ ] <EOL> self . deferred = Deferred ( ) <EOL> self . deferred . addCallback ( lambda r : self . result . append ( r ) ) <EOL> self . deferred . addErrback ( printError ) <EOL> def test_nexus ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = msn . PassportNexus ( self . deferred , '<STR_LIT>' ) <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> transport = StringTransport ( ) <EOL> protocol . makeConnection ( transport ) <EOL> protocol . dataReceived ( '<STR_LIT>' ) <EOL> for ( h , v ) in headers . items ( ) : <EOL> protocol . dataReceived ( '<STR_LIT>' % ( h , v ) ) <EOL> protocol . dataReceived ( '<STR_LIT:\r\n>' ) <EOL> self . assertEqual ( self . result [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> def _doLoginTest ( self , response , headers ) : <EOL> protocol = msn . PassportLogin ( self . deferred , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:a>' ) <EOL> protocol . makeConnection ( StringTransport ( ) ) <EOL> protocol . dataReceived ( response ) <EOL> for ( h , v ) in headers . items ( ) : protocol . dataReceived ( '<STR_LIT>' % ( h , v ) ) <EOL> protocol . dataReceived ( '<STR_LIT:\r\n>' ) <EOL> def testPassportLoginSuccess ( self ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" <EOL> } <EOL> self . _doLoginTest ( '<STR_LIT>' , headers ) <EOL> self . failUnless ( self . result [ <NUM_LIT:0> ] == ( msn . LOGIN_SUCCESS , '<STR_LIT>' ) ) <EOL> def testPassportLoginFailure ( self ) : <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' <EOL> } <EOL> self . _doLoginTest ( '<STR_LIT>' , headers ) <EOL> self . failUnless ( self . result [ <NUM_LIT:0> ] == ( msn . LOGIN_FAILURE , '<STR_LIT>' ) ) <EOL> def testPassportLoginRedirect ( self ) : <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . _doLoginTest ( '<STR_LIT>' , headers ) <EOL> self . failUnless ( self . result [ <NUM_LIT:0> ] == ( msn . LOGIN_REDIRECT , '<STR_LIT>' , '<STR_LIT:a>' ) ) <EOL> if msn is not None : <EOL> class DummySwitchboardClient ( msn . SwitchboardClient ) : <EOL> def userTyping ( self , message ) : <EOL> self . state = '<STR_LIT>' <EOL> def gotSendRequest ( self , fileName , fileSize , cookie , message ) : <EOL> if fileName == '<STR_LIT>' and fileSize == <NUM_LIT> and cookie == <NUM_LIT> : self . state = '<STR_LIT>' <EOL> class DummyNotificationClient ( msn . NotificationClient ) : <EOL> def loggedIn ( self , userHandle , screenName , verified ) : <EOL> if userHandle == '<STR_LIT>' and screenName == '<STR_LIT>' and verified : <EOL> self . state = '<STR_LIT>' <EOL> def gotProfile ( self , message ) : <EOL> self . state = '<STR_LIT>' <EOL> def gotContactStatus ( self , code , userHandle , screenName ) : <EOL> if code == msn . STATUS_AWAY and userHandle == "<STR_LIT>" and screenName == "<STR_LIT>" : <EOL> self . state = '<STR_LIT>' <EOL> def contactStatusChanged ( self , code , userHandle , screenName ) : <EOL> if code == msn . STATUS_LUNCH and userHandle == "<STR_LIT>" and screenName == "<STR_LIT>" : <EOL> self . state = '<STR_LIT>' <EOL> def contactOffline ( self , userHandle ) : <EOL> if userHandle == "<STR_LIT>" : self . state = '<STR_LIT>' <EOL> def statusChanged ( self , code ) : <EOL> if code == msn . STATUS_HIDDEN : self . state = '<STR_LIT>' <EOL> def listSynchronized ( self , * args ) : <EOL> self . state = '<STR_LIT>' <EOL> def gotPhoneNumber ( self , listVersion , userHandle , phoneType , number ) : <EOL> msn . NotificationClient . gotPhoneNumber ( self , listVersion , userHandle , phoneType , number ) <EOL> self . state = '<STR_LIT>' <EOL> def userRemovedMe ( self , userHandle , listVersion ) : <EOL> msn . NotificationClient . userRemovedMe ( self , userHandle , listVersion ) <EOL> c = self . factory . contacts . getContact ( userHandle ) <EOL> if not c and self . factory . contacts . version == listVersion : self . state = '<STR_LIT>' <EOL> def userAddedMe ( self , userHandle , screenName , listVersion ) : <EOL> msn . NotificationClient . userAddedMe ( self , userHandle , screenName , listVersion ) <EOL> c = self . factory . contacts . getContact ( userHandle ) <EOL> if c and ( c . lists | msn . REVERSE_LIST ) and ( self . factory . contacts . version == listVersion ) and ( screenName == '<STR_LIT>' ) : <EOL> self . state = '<STR_LIT>' <EOL> def gotSwitchboardInvitation ( self , sessionID , host , port , key , userHandle , screenName ) : <EOL> if sessionID == <NUM_LIT> and host == '<STR_LIT>' and port == <NUM_LIT> and key == '<STR_LIT>' and userHandle == '<STR_LIT>' and screenName == '<STR_LIT>' : <EOL> self . state = '<STR_LIT>' <EOL> class DispatchTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _versionTest ( self , serverVersionResponse ) : <EOL> """<STR_LIT>""" <EOL> client = msn . DispatchClient ( ) <EOL> client . userHandle = "<STR_LIT:foo>" <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , "<STR_LIT>" ) <EOL> transport . clear ( ) <EOL> client . dataReceived ( serverVersionResponse ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_version ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _versionTest ( "<STR_LIT>" ) <EOL> def test_versionWithoutCVR0 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _versionTest ( "<STR_LIT>" ) <EOL> class NotificationTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . client = DummyNotificationClient ( ) <EOL> self . client . factory = msn . NotificationFactory ( ) <EOL> self . client . state = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> self . client = None <EOL> def _versionTest ( self , serverVersionResponse ) : <EOL> """<STR_LIT>""" <EOL> self . client . factory . userHandle = "<STR_LIT:foo>" <EOL> transport = StringTransport ( ) <EOL> self . client . makeConnection ( transport ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , "<STR_LIT>" ) <EOL> transport . clear ( ) <EOL> self . client . dataReceived ( serverVersionResponse ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_version ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _versionTest ( "<STR_LIT>" ) <EOL> def test_versionWithoutCVR0 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _versionTest ( "<STR_LIT>" ) <EOL> def test_challenge ( self ) : <EOL> """<STR_LIT>""" <EOL> transport = StringTransport ( ) <EOL> self . client . makeConnection ( transport ) <EOL> transport . clear ( ) <EOL> challenge = "<STR_LIT>" <EOL> self . client . dataReceived ( '<STR_LIT>' + challenge + '<STR_LIT:\r\n>' ) <EOL> response = "<STR_LIT>" <EOL> self . assertEqual ( <EOL> response , md5 ( challenge + "<STR_LIT>" ) . hexdigest ( ) ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , <EOL> "<STR_LIT>" + response ) <EOL> def testLogin ( self ) : <EOL> self . client . lineReceived ( '<STR_LIT>' ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def test_loginWithoutSSLFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> self . patch ( msn , '<STR_LIT>' , None ) <EOL> success = [ ] <EOL> self . client . loggedIn = lambda * args : success . append ( args ) <EOL> failure = [ ] <EOL> self . client . loginFailure = failure . append <EOL> self . client . lineReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( success , [ ] ) <EOL> self . assertEqual ( <EOL> failure , <EOL> [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] ) <EOL> def testProfile ( self ) : <EOL> m = '<STR_LIT>' <EOL> m += '<STR_LIT>' <EOL> m += '<STR_LIT>' <EOL> m += '<STR_LIT>' <EOL> map ( self . client . lineReceived , m . split ( '<STR_LIT:\r\n>' ) [ : - <NUM_LIT:1> ] ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def testStatus ( self ) : <EOL> t = [ ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> for i in t : <EOL> self . client . lineReceived ( i [ <NUM_LIT:0> ] ) <EOL> self . failUnless ( ( self . client . state == i [ <NUM_LIT:1> ] ) , msg = i [ <NUM_LIT:2> ] ) <EOL> def testListSync ( self ) : <EOL> self . client . makeConnection ( StringTransport ( ) ) <EOL> msn . NotificationClient . loggedIn ( self . client , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> lines = [ <EOL> "<STR_LIT>" % self . client . currentID , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> map ( self . client . lineReceived , lines ) <EOL> contacts = self . client . factory . contacts <EOL> contact = contacts . getContact ( '<STR_LIT>' ) <EOL> self . failUnless ( contacts . version == <NUM_LIT:100> , "<STR_LIT>" ) <EOL> self . failUnless ( contact . screenName == '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . failUnless ( contacts . groups == { <NUM_LIT:0> : '<STR_LIT>' } , "<STR_LIT>" ) <EOL> self . failUnless ( contact . groups == [ <NUM_LIT:0> ] and contact . lists == <NUM_LIT:11> , "<STR_LIT>" ) <EOL> self . failUnless ( self . client . state == '<STR_LIT>' , "<STR_LIT>" ) <EOL> def testAsyncPhoneChange ( self ) : <EOL> c = msn . MSNContact ( userHandle = '<STR_LIT>' ) <EOL> self . client . factory . contacts = msn . MSNContactList ( ) <EOL> self . client . factory . contacts . addContact ( c ) <EOL> self . client . makeConnection ( StringTransport ( ) ) <EOL> self . client . lineReceived ( "<STR_LIT>" ) <EOL> c = self . client . factory . contacts . getContact ( '<STR_LIT>' ) <EOL> self . failUnless ( self . client . state == '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . failUnless ( c . homePhone == '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . failUnless ( self . client . factory . contacts . version == <NUM_LIT> , "<STR_LIT>" ) <EOL> def testLateBPR ( self ) : <EOL> """<STR_LIT>""" <EOL> self . client . makeConnection ( StringTransport ( ) ) <EOL> msn . NotificationClient . loggedIn ( self . client , '<STR_LIT>' , '<STR_LIT:foo>' , <NUM_LIT:1> ) <EOL> lines = [ <EOL> "<STR_LIT>" % self . client . currentID , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> map ( self . client . lineReceived , lines ) <EOL> contact = self . client . factory . contacts . getContact ( '<STR_LIT>' ) <EOL> self . failUnless ( contact . homePhone == '<STR_LIT>' , "<STR_LIT>" ) <EOL> def testUserRemovedMe ( self ) : <EOL> self . client . factory . contacts = msn . MSNContactList ( ) <EOL> contact = msn . MSNContact ( userHandle = '<STR_LIT>' ) <EOL> contact . addToList ( msn . REVERSE_LIST ) <EOL> self . client . factory . contacts . addContact ( contact ) <EOL> self . client . lineReceived ( "<STR_LIT>" ) <EOL> self . failUnless ( self . client . state == '<STR_LIT>' , "<STR_LIT>" ) <EOL> def testUserAddedMe ( self ) : <EOL> self . client . factory . contacts = msn . MSNContactList ( ) <EOL> self . client . lineReceived ( "<STR_LIT>" ) <EOL> self . failUnless ( self . client . state == '<STR_LIT>' , "<STR_LIT>" ) <EOL> def testAsyncSwitchboardInvitation ( self ) : <EOL> self . client . lineReceived ( "<STR_LIT>" ) <EOL> self . failUnless ( self . client . state == "<STR_LIT>" ) <EOL> def testCommandFailed ( self ) : <EOL> """<STR_LIT>""" <EOL> id , d = self . client . _createIDMapping ( ) <EOL> self . client . lineReceived ( "<STR_LIT>" % id ) <EOL> d = self . assertFailure ( d , msn . MSNCommandFailed ) <EOL> def assertErrorCode ( exception ) : <EOL> self . assertEqual ( <NUM_LIT> , exception . errorCode ) <EOL> return d . addCallback ( assertErrorCode ) <EOL> class MessageHandlingTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . client = DummySwitchboardClient ( ) <EOL> self . client . state = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> self . client = None <EOL> def testClientCapabilitiesCheck ( self ) : <EOL> m = msn . MSNMessage ( ) <EOL> m . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> self . assertEqual ( self . client . checkMessage ( m ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> def testTypingCheck ( self ) : <EOL> m = msn . MSNMessage ( ) <EOL> m . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> m . setHeader ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . client . checkMessage ( m ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def testFileInvitation ( self , lazyClient = False ) : <EOL> m = msn . MSNMessage ( ) <EOL> m . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> m . message += '<STR_LIT>' <EOL> if not lazyClient : <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> self . client . checkMessage ( m ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def testFileInvitationMissingGUID ( self ) : <EOL> return self . testFileInvitation ( True ) <EOL> def testFileResponse ( self ) : <EOL> d = Deferred ( ) <EOL> d . addCallback ( self . fileResponse ) <EOL> self . client . cookies [ '<STR_LIT>' ] [ <NUM_LIT> ] = ( d , None ) <EOL> m = msn . MSNMessage ( ) <EOL> m . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> self . client . checkMessage ( m ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def testFileInfo ( self ) : <EOL> d = Deferred ( ) <EOL> d . addCallback ( self . fileInfo ) <EOL> self . client . cookies [ '<STR_LIT>' ] [ <NUM_LIT> ] = ( d , None ) <EOL> m = msn . MSNMessage ( ) <EOL> m . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> m . message += '<STR_LIT>' <EOL> self . client . checkMessage ( m ) <EOL> self . failUnless ( ( self . client . state == '<STR_LIT>' ) , msg = '<STR_LIT>' ) <EOL> def fileResponse ( self , ( accept , cookie , info ) ) : <EOL> if accept and cookie == <NUM_LIT> : self . client . state = '<STR_LIT>' <EOL> def fileInfo ( self , ( accept , ip , port , aCookie , info ) ) : <EOL> if accept and ip == '<STR_LIT>' and port == <NUM_LIT> and aCookie == <NUM_LIT> : self . client . state = '<STR_LIT>' <EOL> class FileTransferTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . input = '<STR_LIT:a>' * <NUM_LIT> <EOL> self . output = StringIOWithoutClosing ( ) <EOL> def tearDown ( self ) : <EOL> self . input = None <EOL> self . output = None <EOL> def test_fileTransfer ( self ) : <EOL> """<STR_LIT>""" <EOL> auth = <NUM_LIT> <EOL> sender = msn . FileSend ( StringIO . StringIO ( self . input ) ) <EOL> sender . auth = auth <EOL> sender . fileSize = <NUM_LIT> <EOL> client = msn . FileReceive ( auth , "<STR_LIT>" , self . output ) <EOL> client . fileSize = <NUM_LIT> <EOL> def check ( ignored ) : <EOL> self . assertTrue ( <EOL> client . completed and sender . completed , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> self . input , self . output . getvalue ( ) , <EOL> msg = "<STR_LIT>" ) <EOL> d = loopback . loopbackAsync ( sender , client ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> if msn is None : <EOL> for testClass in [ DispatchTests , PassportTests , NotificationTests , <EOL> MessageHandlingTests , FileTransferTestCase ] : <EOL> testClass . skip = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) </s>
