]>
code.delx.au - pymsnt/blob - src/tlib/msn/test_msnw.py
feb78203ea0c656172cac8778923ac4756637663
1 # Copyright 2005 James Bunton <james@delx.cjb.net>
2 # Licensed for distribution under the GPL version 2, check COPYING for details
5 Test cases for msnw (MSN Wrapper)
9 from twisted
. internet
. defer
import Deferred
10 from twisted
. internet
import reactor
, error
11 from twisted
. trial
import unittest
12 from twisted
. python
import log
23 TIMEOUT
= 30.0 # Connection timeout in seconds
25 USER1
= "messengertest1@hotmail.com"
27 USER2
= "messengertest2@hotmail.com"
33 log
. startLogging ( sys
. stdout
)
37 def clearAccount ( msncon
):
38 """ Clears the contact list of the given MSNConnection. Returns a
39 Deferred which fires when the task is complete.
48 if checkCount
== count
:
51 for msnContact
in msncon
. getContacts (). contacts
. values ():
52 for list in [ msn
. FORWARD_LIST
, msn
. BLOCK_LIST
, msn
. ALLOW_LIST
, msn
. PENDING_LIST
]:
53 if msnContact
. lists
& list :
54 msncon
. remContact ( list , msnContact
. userHandle
). addCallback ( cb
)
58 reactor
. callLater ( 0 , d
. callback
, None )
66 class MSNConnection ( msnw
. MSNConnection
):
67 def __init__ ( self
, username
, password
, ident
, testCase
):
68 msnw
. MSNConnection
.__ init
__ ( self
, username
, password
, ident
)
69 self
. testCase
= testCase
71 self
. contactAdded
= None
73 def listSynchronized ( self
):
74 # Now we're fully connected
75 self
. testCase
. done
= "SYNCED"
77 def gotMessage ( self
, userHandle
, text
):
78 self
. testCase
. done
= "GOTMESSAGE"
79 self
. message
= ( userHandle
, text
)
81 def contactAddedMe ( self
, userHandle
):
82 self
. contactAdded
= userHandle
101 def doLogins ( self
, both
= True ):
102 # Connect two accounts
103 self
. user1
= MSNConnection ( USER1
, PASS1
, "user1" , self
)
104 self
. loop ( "Logging in user1." , cond
= "SYNCED" )
106 self
. user2
= MSNConnection ( USER2
, PASS2
, "user2" , self
)
107 self
. loop ( "Logging in user2." , cond
= "SYNCED" )
109 def doPurgeContacts ( self
, both
= True ):
110 # Purge both contact lists
111 clearAccount ( self
. user1
). addCallback ( self
. cb
)
112 self
. loop ( "Purging user1 contact list." )
114 clearAccount ( self
. user2
). addCallback ( self
. cb
)
115 self
. loop ( "Purging user2 contact list." )
117 def doAddContacts ( self
, both
= True ):
118 # Adding users to each other's lists
119 self
. user1
. addContact ( msn
. FORWARD_LIST
, USER2
). addCallback ( self
. cb
)
120 self
. loop ( "Adding user2 to user1's forward list." )
121 self
. user1
. addContact ( msn
. ALLOW_LIST
, USER2
). addCallback ( self
. cb
)
122 self
. loop ( "Adding user2 to user1's allow list." )
124 self
. user2
. addContact ( msn
. FORWARD_LIST
, USER1
). addCallback ( self
. cb
)
125 self
. loop ( "Adding user1 to user2's forward list." )
126 self
. user2
. addContact ( msn
. ALLOW_LIST
, USER1
). addCallback ( self
. cb
)
127 self
. loop ( "Adding user1 to user2's allow list." )
129 # Check the contacts have seen each other
130 reactor
. iterate ( 0.1 ) # One last chance to notice each other
131 self
. failUnless (( self
. user1
. contactAdded
== USER2
and self
. user2
. contactAdded
== USER1
), "Contacts can't see each other." )
133 def cb ( self
, ignored
= None ):
136 def loop ( self
, failMsg
, cond
= True , timeout
= TIMEOUT
):
137 # Loops with a timeout
139 self
. timeout
= reactor
. callLater ( timeout
, self
. failed
, "Timeout: " + failMsg
)
144 while self
. done
!= cond
:
147 self
. timeout
. cancel ()
148 except ( error
. AlreadyCancelled
, error
. AlreadyCalled
):
151 self
. fail ( self
. failure
)
153 self
. failUnless (( self
. done
== cond
), "Failed: " + failMsg
)
155 def failed ( self
, why
):
161 class BasicTests ( unittest
. TestCase
, TestsUtil
):
163 TestsUtil
. setUp ( self
)
166 TestsUtil
. tearDown ( self
)
168 def testConnect ( self
):
169 self
. doLogins ( both
= False )
170 # testConnect.skip = "True"
172 def testPurgeContacts ( self
):
174 self
. doPurgeContacts ()
175 # testPurgeContacts.skip = "True"
177 def testAddContacts ( self
):
179 self
. doPurgeContacts ()
181 # testAddContacts.skip = "True"
183 def testMessageExchange ( self
):
185 self
. doPurgeContacts ()
187 self
. user1
. sendMessage ( USER2
, "Hi user2" )
188 self
. loop ( "Timeout exchanging message." , cond
= "GOTMESSAGE" )
189 self
. failUnless (( self
. user2
. message
== ( USER1
, "Hi user2" )), "Failed to transfer message." )
190 # testMessageExchange.skip = "True"
192 def testFileSend ( self
):
193 if raw_input ( " \n\n ALERT!!! \n\n Please connect to account %s and accept the file transfer from %s . When you have received the complete file, send a message back to the client to signal success. \n Type ok when you are ready: " % ( USER2
, USER1
)). lower () != "ok" :
194 raise unittest
. SkipTest ( "User didn't type 'ok'" )
196 data
= "Testing 123 \r\n " * 5000
197 def accepted (( yes
,)):
199 self
. fileSend
. write ( data
)
200 self
. fileSend
. close ()
202 self
. fail ( "File was not accepted." )
204 self
. fail ( "Transfer failed in invitation." )
205 def gotFileSend (( fileSend
, d
)):
206 self
. fileSend
= fileSend
207 d
. addCallbacks ( accepted
, failed
)
209 self
. doLogins ( both
= False )
210 self
. doPurgeContacts ( both
= False )
211 self
. doAddContacts ( both
= False )
212 d
= self
. user1
. sendFile ( USER2
, "myfile.txt" , len ( data
))
213 d
. addCallback ( gotFileSend
)
214 self
. loop ( "Sending file." , cond
= "GOTMESSAGE" , timeout
= 60 * 60 )
215 testFileSend
. skip
= "True"
217 def testFileReceive ( self
):
218 if raw_input ( " \n\n ALERT!!! \n\n Please connect to account %s and send a file transfer to %s . \n Type ok when you are ready: " % ( USER2
, USER1
)). lower () != "ok" :
219 raise unittest
. SkipTest ( "User didn't type 'ok'" )
221 def fileFinished ( data
):
222 #filename = "/tmp/msn" + str(random.randint(1000, 9999)) + ".dat"
223 filename
= "/tmp/MSNFILE_" + self
. fileReceive
. filename
224 f
= open ( filename
, "w" )
227 print "Got file!" , filename
228 # Terminate the loop in a little, let them send the BYE before
229 # we drop the connection
231 self
. done
= "GOTFILE"
232 reactor
. callLater ( 5 , wait
)
234 def gotFileReceive ( fileReceive
):
235 buffer = msn
. StringBuffer ( fileFinished
)
236 self
. fileReceive
= fileReceive
237 self
. fileReceive
. accept ( buffer )
239 self
. doLogins ( both
= False )
240 self
. user1
. gotFileReceive
= gotFileReceive
241 self
. doPurgeContacts ( both
= False )
242 self
. doAddContacts ( both
= False )
243 self
. loop ( "Receiving file." , cond
= "GOTFILE" , timeout
= 60 * 60 )
244 testFileReceive
. skip
= "True"