]> code.delx.au - pymsnt/blob - src/tlib/msn/test_msn.py
delx.au / code / pymsnt / blob


1fc4ed90b86da00767c76df5b80848a2193e8561
[pymsnt] / src / tlib / msn / test_msn.py
1 # Copyright (c) 2001-2005 Twisted Matrix Laboratories.
2 # Copyright (c) 2005-2006 James Bunton
3 # See LICENSE for details.
4
5 """
6 Test cases for msn.
7 """
8
9 # Settings
10 LOGGING = False
11
12
13
14
15 # Twisted imports
16 from twisted.protocols import loopback
17 from twisted.protocols.basic import LineReceiver
18 from twisted.internet.defer import Deferred
19 from twisted.internet import reactor, main
20 from twisted.python import failure, log
21 from twisted.trial import unittest
22
23 # System imports
24 import StringIO, sys, urllib, random, struct
25
26 import msn
27
28 if LOGGING:
29 log.startLogging(sys.stdout)
30
31
32 def printError(f):
33 print f
34
35 class StringIOWithoutClosing(StringIO.StringIO):
36 disconnecting = 0
37 def close(self): pass
38 def loseConnection(self): pass
39
40 class LoopbackCon:
41 def __init__(self, con1, con2):
42 self.con1 = con1
43 self.con2 = con2
44 self.con1ToCon2 = loopback.LoopbackRelay(con1)
45 self.con2ToCon1 = loopback.LoopbackRelay(con2)
46 con2.makeConnection(self.con1ToCon2)
47 con1.makeConnection(self.con2ToCon1)
48 self.connected = True
49
50 def doSteps(self, steps=1):
51 """ Returns true if the connection finished """
52 count = 0
53 while count < steps:
54 reactor.iterate(0.01)
55 self.con1ToCon2.clearBuffer()
56 self.con2ToCon1.clearBuffer()
57 if self.con1ToCon2.shouldLose:
58 self.con1ToCon2.clearBuffer()
59 count = -1
60 break
61 elif self.con2ToCon1.shouldLose:
62 count = -1
63 break
64 else:
65 count += 1
66 if count == -1:
67 self.disconnect()
68 return True
69 return False
70
71 def disconnect(self):
72 if self.connected:
73 self.con1.connectionLost(failure.Failure(main.CONNECTION_DONE))
74 self.con2.connectionLost(failure.Failure(main.CONNECTION_DONE))
75 reactor.iterate()
76
77
78
79
80 ##################
81 # Passport tests #
82 ##################
83
84 class PassportTests(unittest.TestCase):
85
86 def setUp(self):
87 self.result = []
88 self.deferred = Deferred()
89 self.deferred.addCallback(lambda r: self.result.append(r))
90 self.deferred.addErrback(printError)
91
92 def testNexus(self):
93 protocol = msn.PassportNexus(self.deferred, 'https://foobar.com/somepage.quux')
94 headers = {
95 'Content-Length' : '0',
96 'Content-Type' : 'text/html',
97 'PassportURLs' : 'DARealm=Passport.Net,DALogin=login.myserver.com/,DAReg=reg.myserver.com'
98 }
99 transport = StringIOWithoutClosing()
100 protocol.makeConnection(transport)
101 protocol.dataReceived('HTTP/1.0 200 OK\r\n')
102 for (h,v) in headers.items(): protocol.dataReceived('%s: %s\r\n' % (h,v))
103 protocol.dataReceived('\r\n')
104 self.failUnless(self.result[0] == "https://login.myserver.com/")
105
106 def _doLoginTest(self, response, headers):
107 protocol = msn.PassportLogin(self.deferred,'foo@foo.com','testpass','https://foo.com/', 'a')
108 protocol.makeConnection(StringIOWithoutClosing())
109 protocol.dataReceived(response)
110 for (h,v) in headers.items(): protocol.dataReceived('%s: %s\r\n' % (h,v))
111 protocol.dataReceived('\r\n')
112
113 def testPassportLoginSuccess(self):
114 headers = {
115 'Content-Length' : '0',
116 'Content-Type' : 'text/html',
117 'Authentication-Info' : "Passport1.4 da-status=success,tname=MSPAuth," +
118 "tname=MSPProf,tname=MSPSec,from-PP='somekey'," +
119 "ru=http://messenger.msn.com"
120 }
121 self._doLoginTest('HTTP/1.1 200 OK\r\n', headers)
122 self.failUnless(self.result[0] == (msn.LOGIN_SUCCESS, 'somekey'))
123
124 def testPassportLoginFailure(self):
125 headers = {
126 'Content-Type' : 'text/html',
127 'WWW-Authenticate' : 'Passport1.4 da-status=failed,' +
128 'srealm=Passport.NET,ts=-3,prompt,cburl=http://host.com,' +
129 'cbtxt=the%20error%20message'
130 }
131 self._doLoginTest('HTTP/1.1 401 Unauthorized\r\n', headers)
132 self.failUnless(self.result[0] == (msn.LOGIN_FAILURE, 'the error message'))
133
134 def testPassportLoginRedirect(self):
135 headers = {
136 'Content-Type' : 'text/html',
137 'Authentication-Info' : 'Passport1.4 da-status=redir',
138 'Location' : 'https://newlogin.host.com/'
139 }
140 self._doLoginTest('HTTP/1.1 302 Found\r\n', headers)
141 self.failUnless(self.result[0] == (msn.LOGIN_REDIRECT, 'https://newlogin.host.com/', 'a'))
142
143
144
145 ######################
146 # Notification tests #
147 ######################
148
149 class DummyNotificationClient(msn.NotificationClient):
150 def loggedIn(self, userHandle, verified):
151 if userHandle == 'foo@bar.com' and verified:
152 self.state = 'LOGIN'
153
154 def gotProfile(self, message):
155 self.state = 'PROFILE'
156
157 def gotContactStatus(self, userHandle, code, screenName):
158 if code == msn.STATUS_AWAY and userHandle == "foo@bar.com" and screenName == "Test Screen Name":
159 c = self.factory.contacts.getContact(userHandle)
160 if c.caps & msn.MSNContact.MSNC1 and c.msnobj:
161 self.state = 'INITSTATUS'
162
163 def contactStatusChanged(self, userHandle, code, screenName):
164 if code == msn.STATUS_LUNCH and userHandle == "foo@bar.com" and screenName == "Test Name":
165 self.state = 'NEWSTATUS'
166
167 def contactAvatarChanged(self, userHandle, hash):
168 if userHandle == "foo@bar.com" and hash == "b6b0bc4a5171dac590c593080405921275dcf284":
169 self.state = 'NEWAVATAR'
170 elif self.state == 'NEWAVATAR' and hash == "":
171 self.state = 'AVATARGONE'
172
173 def contactPersonalChanged(self, userHandle, personal):
174 if userHandle == 'foo@bar.com' and personal == 'My Personal Message':
175 self.state = 'GOTPERSONAL'
176 elif userHandle == 'foo@bar.com' and personal == '':
177 self.state = 'PERSONALGONE'
178
179 def contactOffline(self, userHandle):
180 if userHandle == "foo@bar.com": self.state = 'OFFLINE'
181
182 def statusChanged(self, code):
183 if code == msn.STATUS_HIDDEN: self.state = 'MYSTATUS'
184
185 def listSynchronized(self, *args):
186 self.state = 'GOTLIST'
187
188 def gotPhoneNumber(self, userHandle, phoneType, number):
189 self.state = 'GOTPHONE'
190
191 def userRemovedMe(self, userHandle):
192 c = self.factory.contacts.getContact(userHandle)
193 if not c: self.state = 'USERREMOVEDME'
194
195 def userAddedMe(self, userGuid, userHandle, screenName):
196 c = self.factory.contacts.getContact(userHandle)
197 if c and (c.lists | msn.PENDING_LIST) and (screenName == 'Screen Name'):
198 self.state = 'USERADDEDME'
199
200 def gotSwitchboardInvitation(self, sessionID, host, port, key, userHandle, screenName):
201 if sessionID == 1234 and \
202 host == '192.168.1.1' and \
203 port == 1863 and \
204 key == '123.456' and \
205 userHandle == 'foo@foo.com' and \
206 screenName == 'Screen Name':
207 self.state = 'SBINVITED'
208
209 def gotMSNAlert(self, body, action, subscr):
210 self.state = 'NOTIFICATION'
211
212 def gotInitialEmailNotification(self, inboxunread, foldersunread):
213 if inboxunread == 1 and foldersunread == 0:
214 self.state = 'INITEMAIL1'
215 else:
216 self.state = 'INITEMAIL2'
217
218 def gotRealtimeEmailNotification(self, mailfrom, fromaddr, subject):
219 if mailfrom == 'Some Person' and fromaddr == 'example@passport.com' and subject == 'newsubject':
220 self.state = 'REALTIMEEMAIL'
221
222 class NotificationTests(unittest.TestCase):
223 """ testing the various events in NotificationClient """
224
225 def setUp(self):
226 self.client = DummyNotificationClient()
227 self.client.factory = msn.NotificationFactory()
228 self.client.state = 'START'
229
230 def tearDown(self):
231 self.client = None
232
233 def testLogin(self):
234 self.client.lineReceived('USR 1 OK foo@bar.com 1')
235 self.failUnless((self.client.state == 'LOGIN'), 'Failed to detect successful login')
236
237 def testProfile(self):
238 m = 'MSG Hotmail Hotmail 353\r\nMIME-Version: 1.0\r\nContent-Type: text/x-msmsgsprofile; charset=UTF-8\r\n'
239 m += 'LoginTime: 1016941010\r\nEmailEnabled: 1\r\nMemberIdHigh: 40000\r\nMemberIdLow: -600000000\r\nlang_preference: 1033\r\n'
240 m += 'preferredEmail: foo@bar.com\r\ncountry: AU\r\nPostalCode: 90210\r\nGender: M\r\nKid: 0\r\nAge:\r\nsid: 400\r\n'
241 m += 'kv: 2\r\nMSPAuth: 2CACCBCCADMoV8ORoz64BVwmjtksIg!kmR!Rj5tBBqEaW9hc4YnPHSOQ$$\r\n\r\n'
242 self.client.dataReceived(m)
243 self.failUnless((self.client.state == 'PROFILE'), 'Failed to detect initial profile')
244
245 def testInitialEmailNotification(self):
246 m = 'MIME-Version: 1.0\r\nContent-Type: text/x-msmsgsinitialemailnotification; charset=UTF-8\r\n'
247 m += '\r\nInbox-Unread: 1\r\nFolders-Unread: 0\r\nInbox-URL: /cgi-bin/HoTMaiL\r\n'
248 m += 'Folders-URL: /cgi-bin/folders\r\nPost-URL: http://www.hotmail.com\r\n\r\n'
249 m = 'MSG Hotmail Hotmail %s\r\n' % (str(len(m))) + m
250 self.client.dataReceived(m)
251 self.failUnless((self.client.state == 'INITEMAIL1'), 'Failed to detect initial email notification')
252
253 def testNoInitialEmailNotification(self):
254 m = 'MIME-Version: 1.0\r\nContent-Type: text/x-msmsgsinitialemailnotification; charset=UTF-8\r\n'
255 m += '\r\nInbox-Unread: 0\r\nFolders-Unread: 0\r\nInbox-URL: /cgi-bin/HoTMaiL\r\n'
256 m += 'Folders-URL: /cgi-bin/folders\r\nPost-URL: http://www.hotmail.com\r\n\r\n'
257 m = 'MSG Hotmail Hotmail %s\r\n' % (str(len(m))) + m
258 self.client.dataReceived(m)
259 self.failUnless((self.client.state != 'INITEMAIL2'), 'Detected initial email notification when I should not have')
260
261 def testRealtimeEmailNotification(self):
262 m = 'MSG Hotmail Hotmail 356\r\nMIME-Version: 1.0\r\nContent-Type: text/x-msmsgsemailnotification; charset=UTF-8\r\n'
263 m += '\r\nFrom: Some Person\r\nMessage-URL: /cgi-bin/getmsg?msg=MSG1050451140.21&start=2310&len=2059&curmbox=ACTIVE\r\n'
264 m += 'Post-URL: https://loginnet.passport.com/ppsecure/md5auth.srf?lc=1038\r\n'
265 m += 'Subject: =?"us-ascii"?Q?newsubject?=\r\nDest-Folder: ACTIVE\r\nFrom-Addr: example@passport.com\r\nid: 2\r\n'
266 self.client.dataReceived(m)
267 self.failUnless((self.client.state == 'REALTIMEEMAIL'), 'Failed to detect realtime email notification')
268
269 def testMSNAlert(self):
270 m = '<NOTIFICATION ver="2" id="1342902633" siteid="199999999" siteurl="http://alerts.msn.com">\r\n'
271 m += '<TO pid="0x0006BFFD:0x8582C0FB" name="example@passport.com"/>\r\n'
272 m += '<MSG pri="1" id="1342902633">\r\n'
273 m += '<SUBSCR url="http://g.msn.com/3ALMSNTRACKING/199999999ToastChange?http://alerts.msn.com/Alerts/MyAlerts.aspx?strela=1"/>\r\n'
274 m += '<ACTION url="http://g.msn.com/3ALMSNTRACKING/199999999ToastAction?http://alerts.msn.com/Alerts/MyAlerts.aspx?strela=1"/>\r\n'
275 m += '<BODY lang="3076" icon="">\r\n'
276 m += '<TEXT>utf8-encoded text</TEXT></BODY></MSG>\r\n'
277 m += '</NOTIFICATION>\r\n'
278 cmd = 'NOT %s\r\n' % str(len(m))
279 m = cmd + m
280 # Whee, lots of fun to test that lineReceived & dataReceived work well with input coming
281 # in in (fairly) arbitrary chunks.
282 map(self.client.dataReceived, [x+'\r\n' for x in m.split('\r\n')[:-1]])
283 self.failUnless((self.client.state == 'NOTIFICATION'), 'Failed to detect MSN Alert message')
284
285 def testListSync(self):
286 self.client.makeConnection(StringIOWithoutClosing())
287 msn.NotificationClient.loggedIn(self.client, 'foo@foo.com', 1)
288 lines = [
289 "SYN %s 2005-04-23T18:57:44.8130000-07:00 2005-04-23T18:57:54.2070000-07:00 1 3" % self.client.currentID,
290 "GTC A",
291 "BLP AL",
292 "LSG Friends yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy",
293 "LSG Other%20Friends yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyz",
294 "LSG More%20Other%20Friends yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyya",
295 "LST N=userHandle@email.com F=Some%20Name C=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx 13 yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy,yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyz",
296 ]
297 map(self.client.lineReceived, lines)
298 contacts = self.client.factory.contacts
299 contact = contacts.getContact('userHandle@email.com')
300 #self.failUnless(contacts.version == 100, "Invalid contact list version")
301 self.failUnless(contact.screenName == 'Some Name', "Invalid screen-name for user")
302 self.failUnless(contacts.groups == {'yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy': 'Friends', \
303 'yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyz': 'Other Friends', \
304 'yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyya': 'More Other Friends'} \
305 , "Did not get proper group list")
306 self.failUnless(contact.groups == ['yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy', \
307 'yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyz'] and \
308 contact.lists == 13, "Invalid contact list/group info")
309 self.failUnless(self.client.state == 'GOTLIST', "Failed to call list sync handler")
310 self.client.logOut()
311
312 def testStatus(self):
313 # Set up the contact list
314 self.client.makeConnection(StringIOWithoutClosing())
315 msn.NotificationClient.loggedIn(self.client, 'foo@foo.com', 1)
316 lines = [
317 "SYN %s 2005-04-23T18:57:44.8130000-07:00 2005-04-23T18:57:54.2070000-07:00 1 0" % self.client.currentID,
318 "GTC A",
319 "BLP AL",
320 "LST N=foo@bar.com F=Some%20Name C=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx 13 yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy,yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyz",
321 ]
322 map(self.client.lineReceived, lines)
323 # Now test!
324 msnobj = urllib.quote('<msnobj Creator="buddy1@hotmail.com" Size="24539" Type="3" Location="TFR2C.tmp" Friendly="AAA=" SHA1D="trC8SlFx2sWQxZMIBAWSEnXc8oQ=" SHA1C="U32o6bosZzluJq82eAtMpx5dIEI="/>')
325 t = [('ILN 1 AWY foo@bar.com Test%20Screen%20Name 268435456 ' + msnobj, 'INITSTATUS', 'Failed to detect initial status report'),
326 ('NLN LUN foo@bar.com Test%20Name 0', 'NEWSTATUS', 'Failed to detect contact status change'),
327 ('NLN AWY foo@bar.com Test%20Name 0 ' + msnobj, 'NEWAVATAR', 'Failed to detect contact avatar change'),
328 ('NLN AWY foo@bar.com Test%20Name 0', 'AVATARGONE', 'Failed to detect contact avatar disappearing'),
329 ('FLN foo@bar.com', 'OFFLINE', 'Failed to detect contact signing off'),
330 ('CHG 1 HDN 0 ' + msnobj, 'MYSTATUS', 'Failed to detect my status changing')]
331 for i in t:
332 self.client.lineReceived(i[0])
333 self.failUnless((self.client.state == i[1]), i[2])
334
335 # Test UBX
336 self.client.dataReceived('UBX foo@bar.com 72\r\n<Data><PSM>My Personal Message</PSM><CurrentMedia></CurrentMedia></Data>')
337 self.failUnless((self.client.state == 'GOTPERSONAL'), 'Failed to detect new personal message')
338 self.client.dataReceived('UBX foo@bar.com 0\r\n')
339 self.failUnless((self.client.state == 'PERSONALGONE'), 'Failed to detect personal message disappearing')
340 self.client.logOut()
341
342 def testAsyncPhoneChange(self):
343 c = msn.MSNContact(userHandle='userHandle@email.com')
344 self.client.factory.contacts = msn.MSNContactList()
345 self.client.factory.contacts.addContact(c)
346 self.client.makeConnection(StringIOWithoutClosing())
347 self.client.lineReceived("BPR 101 userHandle@email.com PHH 123%20456")
348 c = self.client.factory.contacts.getContact('userHandle@email.com')
349 self.failUnless(self.client.state == 'GOTPHONE', "Did not fire phone change callback")
350 self.failUnless(c.homePhone == '123 456', "Did not update the contact's phone number")
351 self.failUnless(self.client.factory.contacts.version == 101, "Did not update list version")
352
353 def testLateBPR(self):
354 """
355 This test makes sure that if a BPR response that was meant
356 to be part of a SYN response (but came after the last LST)
357 is received, the correct contact is updated and all is well
358 """
359 self.client.makeConnection(StringIOWithoutClosing())
360 msn.NotificationClient.loggedIn(self.client, 'foo@foo.com', 1)
361 lines = [
362 "SYN %s 2005-04-23T18:57:44.8130000-07:00 2005-04-23T18:57:54.2070000-07:00 1 0" % self.client.currentID,
363 "GTC A",
364 "BLP AL",
365 "LSG Friends yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy",
366 "LST N=userHandle@email.com F=Some%20Name C=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx 13 yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy",
367 "BPR PHH 123%20456"
368 ]
369 map(self.client.lineReceived, lines)
370 contact = self.client.factory.contacts.getContact('userHandle@email.com')
371 self.failUnless(contact.homePhone == '123 456', "Did not update contact's phone number")
372 self.client.logOut()
373
374 def testUserRemovedMe(self):
375 self.client.factory.contacts = msn.MSNContactList()
376 contact = msn.MSNContact(userHandle='foo@foo.com')
377 contact.addToList(msn.REVERSE_LIST)
378 self.client.factory.contacts.addContact(contact)
379 self.client.lineReceived("REM 0 RL foo@foo.com")
380 self.failUnless(self.client.state == 'USERREMOVEDME', "Failed to remove user from reverse list")
381
382 def testUserAddedMe(self):
383 self.client.factory.contacts = msn.MSNContactList()
384 self.client.lineReceived("ADC 0 RL N=foo@foo.com F=Screen%20Name")
385 self.failUnless(self.client.state == 'USERADDEDME', "Failed to add user to reverse lise")
386
387 def testAsyncSwitchboardInvitation(self):
388 self.client.lineReceived("RNG 1234 192.168.1.1:1863 CKI 123.456 foo@foo.com Screen%20Name")
389 self.failUnless((self.client.state == 'SBINVITED'), 'Failed to detect switchboard invitation')
390
391
392 #######################################
393 # Notification with fake server tests #
394 #######################################
395
396 class FakeNotificationServer(msn.MSNEventBase):
397 def handle_CHG(self, params):
398 if len(params) < 4:
399 params.append('')
400 self.sendLine("CHG %s %s %s %s" % (params[0], params[1], params[2], params[3]))
401
402 def handle_BLP(self, params):
403 self.sendLine("BLP %s %s 100" % (params[0], params[1]))
404
405 def handle_ADC(self, params):
406 trid = params[0]
407 list = msn.listCodeToID[params[1].lower()]
408 if list == msn.FORWARD_LIST:
409 userHandle = ""
410 screenName = ""
411 userGuid = ""
412 groups = ""
413 for p in params[2:]:
414 if p[0] == 'N':
415 userHandle = p[2:]
416 elif p[0] == 'F':
417 screenName = p[2:]
418 elif p[0] == 'C':
419 userGuid = p[2:]
420 else:
421 groups = p
422 if userHandle and userGuid:
423 self.transport.loseConnection()
424 return
425 if userHandle:
426 if not screenName:
427 self.transport.loseConnection()
428 else:
429 self.sendLine("ADC %s FL N=%s F=%s C=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx %s" % (trid, userHandle, screenName, groups))
430 return
431 if userGuid:
432 raise "NotImplementedError"
433 else:
434 if len(params) != 3:
435 self.transport.loseConnection()
436 if not params[2].startswith("N=") and params[2].count('@') == 1:
437 self.transport.loseConnection()
438 self.sendLine("ADC %s %s %s" % (params[0], params[1], params[2]))
439
440 def handle_REM(self, params):
441 if len(params) != 3:
442 self.transport.loseConnection()
443 return
444 try:
445 trid = int(params[0])
446 listType = msn.listCodeToID[params[1].lower()]
447 except:
448 self.transport.loseConnection()
449 if listType == msn.FORWARD_LIST and params[2].count('@') > 0:
450 self.transport.loseConnection()
451 elif listType != msn.FORWARD_LIST and params[2].count('@') != 1:
452 self.transport.loseConnection()
453 else:
454 self.sendLine("REM %s %s %s" % (params[0], params[1], params[2]))
455
456 def handle_PRP(self, params):
457 if len(params) != 3:
458 self.transport.loseConnection()
459 if params[1] == "MFN":
460 self.sendLine("PRP %s MFN %s" % (params[0], params[2]))
461 else:
462 # Only friendly names are implemented
463 self.transport.loseConnection()
464
465 def handle_UUX(self, params):
466 if len(params) != 2:
467 self.transport.loseConnection()
468 return
469 l = int(params[1])
470 if l > 0:
471 self.currentMessage = msn.MSNMessage(length=l, userHandle=params[0], screenName="UUX", specialMessage=True)
472 self.setRawMode()
473 else:
474 self.sendLine("UUX %s 0" % params[0])
475
476 def checkMessage(self, message):
477 if message.specialMessage:
478 if message.screenName == "UUX":
479 self.sendLine("UUX %s 0" % message.userHandle)
480 return 0
481 return 1
482
483 def handle_XFR(self, params):
484 if len(params) != 2:
485 self.transport.loseConnection()
486 return
487 if params[1] != "SB":
488 self.transport.loseConnection()
489 return
490 self.sendLine("XFR %s SB 129.129.129.129:1234 CKI SomeSecret" % params[0])
491
492
493
494 class FakeNotificationClient(msn.NotificationClient):
495 def doStatusChange(self):
496 def testcb((status,)):
497 if status == msn.STATUS_AWAY:
498 self.test = 'PASS'
499 self.transport.loseConnection()
500 d = self.changeStatus(msn.STATUS_AWAY)
501 d.addCallback(testcb)
502
503 def doPrivacyMode(self):
504 def testcb((priv,)):
505 if priv.upper() == 'AL':
506 self.test = 'PASS'
507 self.transport.loseConnection()
508 d = self.setPrivacyMode(True)
509 d.addCallback(testcb)
510
511 def doAddContactFL(self):
512 def testcb((listType, userGuid, userHandle, screenName)):
513 if listType & msn.FORWARD_LIST and \
514 userGuid == "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" and \
515 userHandle == "foo@bar.com" and \
516 screenName == "foo@bar.com" and \
517 self.factory.contacts.getContact(userHandle):
518 self.test = 'PASS'
519 self.transport.loseConnection()
520 d = self.addContact(msn.FORWARD_LIST, "foo@bar.com")
521 d.addCallback(testcb)
522
523 def doAddContactAL(self):
524 def testcb((listType, userGuid, userHandle, screenName)):
525 if listType & msn.ALLOW_LIST and \
526 userHandle == "foo@bar.com" and \
527 not userGuid and not screenName and \
528 self.factory.contacts.getContact(userHandle):
529 self.test = 'PASS'
530 self.transport.loseConnection()
531 d = self.addContact(msn.ALLOW_LIST, "foo@bar.com")
532 d.addCallback(testcb)
533
534 def doRemContactFL(self):
535 def testcb((listType, userHandle, groupID)):
536 if listType & msn.FORWARD_LIST and \
537 userHandle == "foo@bar.com":
538 self.test = 'PASS'
539 self.transport.loseConnection()
540 d = self.remContact(msn.FORWARD_LIST, "foo@bar.com")
541 d.addCallback(testcb)
542
543 def doRemContactAL(self):
544 def testcb((listType, userHandle, groupID)):
545 if listType & msn.ALLOW_LIST and \
546 userHandle == "foo@bar.com":
547 self.test = 'PASS'
548 self.transport.loseConnection()
549 d = self.remContact(msn.ALLOW_LIST, "foo@bar.com")
550 d.addCallback(testcb)
551
552 def doScreenNameChange(self):
553 def testcb(*args):
554 self.test = 'PASS'
555 self.transport.loseConnection()
556 d = self.changeScreenName("Some new name")
557 d.addCallback(testcb)
558
559 def doPersonalChange(self, personal):
560 def testcb((checkPersonal,)):
561 if checkPersonal == personal:
562 self.test = 'PASS'
563 self.transport.loseConnection()
564 d = self.changePersonalMessage(personal)
565 d.addCallback(testcb)
566
567 def doAvatarChange(self, data):
568 def testcb(ignored):
569 self.test = 'PASS'
570 self.transport.loseConnection()
571 d = self.changeAvatar(data, True)
572 d.addCallback(testcb)
573
574 def doRequestSwitchboard(self):
575 def testcb((host, port, key)):
576 if host == "129.129.129.129" and port == 1234 and key == "SomeSecret":
577 self.test = 'PASS'
578 self.transport.loseConnection()
579 d = self.requestSwitchboardServer()
580 d.addCallback(testcb)
581
582 class FakeServerNotificationTests(unittest.TestCase):
583 """ tests the NotificationClient against a fake server. """
584
585 def setUp(self):
586 self.client = FakeNotificationClient()
587 self.client.factory = msn.NotificationFactory()
588 self.client.test = 'FAIL'
589 self.server = FakeNotificationServer()
590 self.loop = LoopbackCon(self.client, self.server)
591
592 def tearDown(self):
593 self.loop.disconnect()
594
595 def testChangeStatus(self):
596 self.client.doStatusChange()
597 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
598 self.failUnless((self.client.test == 'PASS'), 'Failed to change status properly')
599
600 def testSetPrivacyMode(self):
601 self.client.factory.contacts = msn.MSNContactList()
602 self.client.doPrivacyMode()
603 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
604 self.failUnless((self.client.test == 'PASS'), 'Failed to change privacy mode')
605
606 def testSyncList(self):
607 self.client.doSyncList()
608 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
609 self.failUnless((self.client.test == 'PASS'), 'Failed to synchronise list')
610 testSyncList.skip = "Will do after list versions."
611
612 def testAddContactFL(self):
613 self.client.factory.contacts = msn.MSNContactList()
614 self.client.doAddContactFL()
615 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
616 self.failUnless((self.client.test == 'PASS'), 'Failed to add contact to forward list')
617
618 def testAddContactAL(self):
619 self.client.factory.contacts = msn.MSNContactList()
620 self.client.doAddContactAL()
621 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
622 self.failUnless((self.client.test == 'PASS'), 'Failed to add contact to allow list')
623
624 def testRemContactFL(self):
625 self.client.factory.contacts = msn.MSNContactList()
626 self.client.factory.contacts.addContact(msn.MSNContact(userGuid="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", userHandle="foo@bar.com", screenName="Some guy", lists=msn.FORWARD_LIST))
627 self.client.doRemContactFL()
628 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
629 self.failUnless((self.client.test == 'PASS'), 'Failed to remove contact from forward list')
630
631 def testRemContactAL(self):
632 self.client.factory.contacts = msn.MSNContactList()
633 self.client.factory.contacts.addContact(msn.MSNContact(userGuid="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", userHandle="foo@bar.com", screenName="Some guy", lists=msn.ALLOW_LIST))
634 self.client.doRemContactAL()
635 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
636 self.failUnless((self.client.test == 'PASS'), 'Failed to remove contact from allow list')
637
638 def testChangedScreenName(self):
639 self.client.doScreenNameChange()
640 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
641 self.failUnless((self.client.test == 'PASS'), 'Failed to change screen name properly')
642
643 def testChangePersonal1(self):
644 self.client.doPersonalChange("Some personal message")
645 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
646 self.failUnless((self.client.test == 'PASS'), 'Failed to change personal message properly')
647
648 def testChangePersonal2(self):
649 self.client.doPersonalChange("")
650 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
651 self.failUnless((self.client.test == 'PASS'), 'Failed to change personal message properly')
652
653 def testChangeAvatar(self):
654 self.client.doAvatarChange("DATADATADATADATA")
655 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
656 self.failUnless((self.client.test == 'PASS'), 'Failed to change avatar properly')
657
658 def testRequestSwitchboard(self):
659 self.client.doRequestSwitchboard()
660 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
661 self.failUnless((self.client.test == 'PASS'), 'Failed to request switchboard')
662
663
664 #################################
665 # Notification challenges tests #
666 #################################
667
668 class DummyChallengeNotificationServer(msn.MSNEventBase):
669 def doChallenge(self, challenge, response):
670 self.state = 0
671 self.response = response
672 self.sendLine("CHL 0 " + challenge)
673
674 def checkMessage(self, message):
675 if message.message == self.response:
676 self.state = "PASS"
677 self.transport.loseConnection()
678 return 0
679
680 def handle_QRY(self, params):
681 self.state = 1
682 if len(params) == 3 and params[1] == "PROD0090YUAUV{2B" and params[2] == "32":
683 self.state = 2
684 self.currentMessage = msn.MSNMessage(length=32, userHandle="QRY", screenName="QRY", specialMessage=True)
685 self.setRawMode()
686 else:
687 self.transport.loseConnection()
688
689 class DummyChallengeNotificationClient(msn.NotificationClient):
690 def connectionMade(self):
691 msn.MSNEventBase.connectionMade(self)
692
693 def handle_CHL(self, params):
694 msn.NotificationClient.handle_CHL(self, params)
695 self.transport.loseConnection()
696
697
698 class NotificationChallengeTests(unittest.TestCase):
699 """ tests the responses to the CHLs the server sends """
700
701 def setUp(self):
702 self.client = DummyChallengeNotificationClient()
703 self.server = DummyChallengeNotificationServer()
704 self.loop = LoopbackCon(self.client, self.server)
705
706 def tearDown(self):
707 self.loop.disconnect()
708
709 def testChallenges(self):
710 challenges = [('13038318816579321232', 'b01c13020e374d4fa20abfad6981b7a9'),
711 ('23055170411503520698', 'ae906c3f2946d25e7da1b08b0b247659'),
712 ('37819769320541083311', 'db79d37dadd9031bef996893321da480'),
713 ('93662730714769834295', 'd619dfbb1414004d34d0628766636568'),
714 ('31154116582196216093', '95e96c4f8cfdba6f065c8869b5e984e9')]
715 for challenge, response in challenges:
716 self.server.doChallenge(challenge, response)
717 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
718 self.failUnless((self.server.state == 'PASS'), 'Incorrect challenge response.')
719
720
721 ###########################
722 # Notification ping tests #
723 ###########################
724
725 class DummyPingNotificationServer(LineReceiver):
726 def lineReceived(self, line):
727 if line.startswith("PNG") and self.good:
728 self.sendLine("QNG 50")
729
730 class DummyPingNotificationClient(msn.NotificationClient):
731 def connectionMade(self):
732 self.pingCheckerStart()
733
734 def sendLine(self, line):
735 msn.NotificationClient.sendLine(self, line)
736 self.count += 1
737 if self.count > 10:
738 self.transport.loseConnection() # But not for real, just to end the test
739
740 def connectionLost(self, reason):
741 if self.count <= 10:
742 self.state = 'DISCONNECTED'
743
744 class NotificationPingTests(unittest.TestCase):
745 """ tests pinging in the NotificationClient class """
746
747 def setUp(self):
748 msn.PINGSPEED = 0.1
749 self.client = DummyPingNotificationClient()
750 self.server = DummyPingNotificationServer()
751 self.client.state = 'CONNECTED'
752 self.client.count = 0
753 self.loop = LoopbackCon(self.client, self.server)
754
755 def tearDown(self):
756 msn.PINGSPEED = 50.0
757 self.client.logOut()
758 self.loop.disconnect()
759
760 def testPingGood(self):
761 self.server.good = True
762 self.loop.doSteps(100)
763 self.failUnless((self.client.state == 'CONNECTED'), 'Should be connected.')
764
765 def testPingBad(self):
766 self.server.good = False
767 self.loop.doSteps(100)
768 self.failUnless((self.client.state == 'DISCONNECTED'), 'Should be disconnected.')
769
770
771
772
773 ###########################
774 # Switchboard basic tests #
775 ###########################
776
777 class DummySwitchboardServer(msn.MSNEventBase):
778 def handle_USR(self, params):
779 if len(params) != 3:
780 self.transport.loseConnection()
781 if params[1] == 'foo@bar.com' and params[2] == 'somekey':
782 self.sendLine("USR %s OK %s %s" % (params[0], params[1], params[1]))
783
784 def handle_ANS(self, params):
785 if len(params) != 4:
786 self.transport.loseConnection()
787 if params[1] == 'foo@bar.com' and params[2] == 'somekey' and params[3] == 'someSID':
788 self.sendLine("ANS %s OK" % params[0])
789
790 def handle_CAL(self, params):
791 if len(params) != 2:
792 self.transport.loseConnection()
793 if params[1] == 'friend@hotmail.com':
794 self.sendLine("CAL %s RINGING 1111122" % params[0])
795 else:
796 self.transport.loseConnection()
797
798 def checkMessage(self, message):
799 if message.message == 'Hi how are you today?':
800 self.sendLine("ACK " + message.userHandle) # Relies on TRID getting stored in userHandle trick
801 else:
802 self.transport.loseConnection()
803 return 0
804
805 class DummySwitchboardClient(msn.SwitchboardClient):
806 def loggedIn(self):
807 self.state = 'LOGGEDIN'
808 self.transport.loseConnection()
809
810 def gotChattingUsers(self, users):
811 if users == {'fred@hotmail.com': 'fred', 'jack@email.com': 'jack has a nickname!'}:
812 self.state = 'GOTCHATTINGUSERS'
813
814 def userJoined(self, userHandle, screenName):
815 if userHandle == "friend@hotmail.com" and screenName == "friend nickname":
816 self.state = 'USERJOINED'
817
818 def userLeft(self, userHandle):
819 if userHandle == "friend@hotmail.com":
820 self.state = 'USERLEFT'
821
822 def gotContactTyping(self, message):
823 if message.userHandle == 'foo@bar.com':
824 self.state = 'USERTYPING'
825
826 def gotMessage(self, message):
827 if message.userHandle == 'friend@hotmail.com' and \
828 message.screenName == 'Friend Nickname' and \
829 message.message == 'Hello.':
830 self.state = 'GOTMESSAGE'
831
832 def doSendInvite(self):
833 def testcb((sid,)):
834 if sid == 1111122:
835 self.state = 'INVITESUCCESS'
836 self.transport.loseConnection()
837 d = self.inviteUser('friend@hotmail.com')
838 d.addCallback(testcb)
839
840 def doSendMessage(self):
841 def testcb(ignored):
842 self.state = 'MESSAGESUCCESS'
843 self.transport.loseConnection()
844 m = msn.MSNMessage()
845 m.setHeader("Content-Type", "text/plain; charset=UTF-8")
846 m.message = 'Hi how are you today?'
847 m.ack = msn.MSNMessage.MESSAGE_ACK
848 d = self.sendMessage(m)
849 d.addCallback(testcb)
850
851
852 class SwitchboardBasicTests(unittest.TestCase):
853 """ Tests basic functionality of switchboard sessions """
854 def setUp(self):
855 self.client = DummySwitchboardClient()
856 self.client.state = 'START'
857 self.client.userHandle = 'foo@bar.com'
858 self.client.key = 'somekey'
859 self.client.sessionID = 'someSID'
860 self.server = DummySwitchboardServer()
861 self.loop = LoopbackCon(self.client, self.server)
862
863 def tearDown(self):
864 self.loop.disconnect()
865
866 def _testSB(self, reply):
867 self.client.reply = reply
868 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
869 self.failUnless((self.client.state == 'LOGGEDIN'), 'Failed to login with reply='+str(reply))
870
871 def testReply(self):
872 self._testSB(True)
873
874 def testAsync(self):
875 self._testSB(False)
876
877 def testChattingUsers(self):
878 lines = ["IRO 1 1 2 fred@hotmail.com fred",
879 "IRO 1 2 2 jack@email.com jack%20has%20a%20nickname%21"]
880 for line in lines:
881 self.client.lineReceived(line)
882 self.failUnless((self.client.state == 'GOTCHATTINGUSERS'), 'Failed to get chatting users')
883
884 def testUserJoined(self):
885 self.client.lineReceived("JOI friend@hotmail.com friend%20nickname")
886 self.failUnless((self.client.state == 'USERJOINED'), 'Failed to notice user joining')
887
888 def testUserLeft(self):
889 self.client.lineReceived("BYE friend@hotmail.com")
890 self.failUnless((self.client.state == 'USERLEFT'), 'Failed to notice user leaving')
891
892 def testTypingCheck(self):
893 m = 'MSG foo@bar.com Foo 80\r\n'
894 m += 'MIME-Version: 1.0\r\n'
895 m += 'Content-Type: text/x-msmsgscontrol\r\n'
896 m += 'TypingUser: foo@bar\r\n'
897 m += '\r\n\r\n'
898 self.client.dataReceived(m)
899 self.failUnless((self.client.state == 'USERTYPING'), 'Failed to detect typing notification')
900
901 def testGotMessage(self):
902 m = 'MSG friend@hotmail.com Friend%20Nickname 68\r\n'
903 m += 'MIME-Version: 1.0\r\n'
904 m += 'Content-Type: text/plain; charset=UTF-8\r\n'
905 m += '\r\nHello.'
906 self.client.dataReceived(m)
907 self.failUnless((self.client.state == 'GOTMESSAGE'), 'Failed to detect message')
908
909 def testInviteUser(self):
910 self.client.connectionMade = lambda: None
911 self.client.doSendInvite()
912 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
913 self.failUnless((self.client.state == 'INVITESUCCESS'), 'Failed to invite user')
914
915 def testSendMessage(self):
916 self.client.connectionMade = lambda: None
917 self.client.doSendMessage()
918 self.failUnless(self.loop.doSteps(10), 'Failed to disconnect')
919 self.failUnless((self.client.state == 'MESSAGESUCCESS'), 'Failed to send message')
920
921
922 ################
923 # MSNP2P tests #
924 ################
925
926 class DummySwitchboardP2PServerHelper(msn.MSNEventBase):
927 def __init__(self, server):
928 msn.MSNEventBase.__init__(self)
929 self.server = server
930
931 def handle_USR(self, params):
932 if len(params) != 3:
933 self.transport.loseConnection()
934 self.userHandle = params[1]
935 if params[1] == 'foo1@bar.com' and params[2] == 'somekey1':
936 self.sendLine("USR %s OK %s %s" % (params[0], params[1], params[1]))
937 if params[1] == 'foo2@bar.com' and params[2] == 'somekey2':
938 self.sendLine("USR %s OK %s %s" % (params[0], params[1], params[1]))
939
940 def checkMessage(self, message):
941 return 1
942
943 def gotMessage(self, message):
944 message.userHandle = self.userHandle
945 message.screenName = self.userHandle
946 self.server.gotMessage(message, self)
947
948 def sendMessage(self, message):
949 if message.length == 0: message.length = message._calcMessageLen()
950 self.sendLine("MSG %s %s %s" % (message.userHandle, message.screenName, message.length))
951 self.sendLine('MIME-Version: %s' % message.getHeader('MIME-Version'))
952 self.sendLine('Content-Type: %s' % message.getHeader('Content-Type'))
953 for header in [h for h in message.headers.items() if h[0].lower() not in ('mime-version','content-type')]:
954 self.sendLine("%s: %s" % (header[0], header[1]))
955 self.transport.write("\r\n")
956 self.transport.write(message.message)
957
958
959 class DummySwitchboardP2PServer:
960 def __init__(self):
961 self.clients = []
962
963 def newClient(self):
964 c = DummySwitchboardP2PServerHelper(self)
965 self.clients.append(c)
966 return c
967
968 def gotMessage(self, message, sender):
969 for c in self.clients:
970 if c != sender:
971 c.sendMessage(message)
972
973 class DummySwitchboardP2PClient(msn.SwitchboardClient):
974 def gotMessage(self, message):
975 if message.message == "Test Message" and message.userHandle == "foo1@bar.com":
976 self.status = "GOTMESSAGE"
977
978 def gotFileReceive(self, fileReceive):
979 self.fileReceive = fileReceive
980
981 class SwitchboardP2PTests(unittest.TestCase):
982 def setUp(self):
983 self.server = DummySwitchboardP2PServer()
984 self.client1 = DummySwitchboardP2PClient()
985 self.client1.key = 'somekey1'
986 self.client1.userHandle = 'foo1@bar.com'
987 self.client2 = DummySwitchboardP2PClient()
988 self.client2.key = 'somekey2'
989 self.client2.userHandle = 'foo2@bar.com'
990 self.client2.status = "INIT"
991 self.loop1 = LoopbackCon(self.client1, self.server.newClient())
992 self.loop2 = LoopbackCon(self.client2, self.server.newClient())
993
994 def tearDown(self):
995 self.loop1.disconnect()
996 self.loop2.disconnect()
997
998 def _loop(self, steps=1):
999 for i in xrange(steps):
1000 self.loop1.doSteps(1)
1001 self.loop2.doSteps(1)
1002
1003 def testMessage(self):
1004 self.client1.sendMessage(msn.MSNMessage(message='Test Message'))
1005 self._loop()
1006 self.failUnless((self.client2.status == "GOTMESSAGE"), "Fake switchboard server not working.")
1007
1008 def _generateData(self):
1009 data = ''
1010 for i in xrange(3000):
1011 data += struct.pack("<L", random.randint(0, sys.maxint))
1012 return data
1013
1014 def testAvatars(self):
1015 self.gotAvatar = False
1016
1017 # Set up the avatar for client1
1018 imageData = self._generateData()
1019 self.client1.msnobj = msn.MSNObject()
1020 self.client1.msnobj.setData('foo1@bar.com', imageData)
1021 self.client1.msnobj.makeText()
1022
1023 # Make client2 request the avatar
1024 def avatarCallback((data,)):
1025 self.gotAvatar = (data == imageData)
1026 msnContact = msn.MSNContact(userHandle='foo1@bar.com', msnobj=self.client1.msnobj)
1027 d = self.client2.sendAvatarRequest(msnContact)
1028 d.addCallback(avatarCallback)
1029
1030 # Let them do their thing
1031 self._loop(10)
1032
1033 # Check that client2 got the avatar
1034 self.failUnless((self.gotAvatar), "Failed to transfer avatar")
1035
1036 def testFilesHappyPath(self):
1037 fileData = self._generateData()
1038 self.gotFile = False
1039
1040 # Send the file (client2->client1)
1041 msnContact = msn.MSNContact(userHandle='foo1@bar.com', caps=msn.MSNContact.MSNC1)
1042 fileSend, d = self.client2.sendFile(msnContact, "myfile.txt", len(fileData))
1043 def accepted((yes,)):
1044 if yes:
1045 fileSend.write(fileData)
1046 fileSend.close()
1047 else:
1048 raise "TransferDeclined"
1049 def failed():
1050 raise "TransferError"
1051 d.addCallback(accepted)
1052 d.addErrback(failed)
1053
1054 # Let the request get pushed to client1
1055 self._loop(10)
1056
1057 # Receive the file
1058 def finished(data):
1059 self.gotFile = (data == fileData)
1060 fileBuffer = msn.StringBuffer(finished)
1061 fileReceive = self.client1.fileReceive
1062 self.failUnless((fileReceive.filename == "myfile.txt" and fileReceive.filesize == len(fileData)), "Filename or length wrong.")
1063 fileReceive.accept(fileBuffer)
1064
1065 # Lets transfer!!
1066 self._loop(10)
1067
1068 self.failUnless((self.gotFile), "Failed to transfer file")
1069
1070 def testFilesHappyChunkedPath(self):
1071 fileData = self._generateData()
1072 self.gotFile = False
1073
1074 # Send the file (client2->client1)
1075 msnContact = msn.MSNContact(userHandle='foo1@bar.com', caps=msn.MSNContact.MSNC1)
1076 fileSend, d = self.client2.sendFile(msnContact, "myfile.txt", len(fileData))
1077 def accepted((yes,)):
1078 if yes:
1079 fileSend.write(fileData[:len(fileData)/2])
1080 fileSend.write(fileData[len(fileData)/2:])
1081 fileSend.close()
1082 else:
1083 raise "TransferDeclined"
1084 def failed():
1085 raise "TransferError"
1086 d.addCallback(accepted)
1087 d.addErrback(failed)
1088
1089 # Let the request get pushed to client1
1090 self._loop(10)
1091
1092 # Receive the file
1093 def finished(data):
1094 self.gotFile = (data == fileData)
1095 fileBuffer = msn.StringBuffer(finished)
1096 fileReceive = self.client1.fileReceive
1097 self.failUnless((fileReceive.filename == "myfile.txt" and fileReceive.filesize == len(fileData)), "Filename or length wrong.")
1098 fileReceive.accept(fileBuffer)
1099
1100 # Lets transfer!!
1101 self._loop(10)
1102
1103 self.failUnless((self.gotFile), "Failed to transfer file")
1104
1105 def testTwoFilesSequential(self):
1106 self.testFilesHappyPath()
1107 self.testFilesHappyPath()
1108
1109 def testFilesDeclinePath(self):
1110 fileData = self._generateData()
1111 self.gotDecline = False
1112
1113 # Send the file (client2->client1)
1114 msnContact = msn.MSNContact(userHandle='foo1@bar.com', caps=msn.MSNContact.MSNC1)
1115 fileSend, d = self.client2.sendFile(msnContact, "myfile.txt", len(fileData))
1116 def accepted((yes,)):
1117 self.failUnless((not yes), "Failed to understand a decline.")
1118 self.gotDecline = True
1119 def failed():
1120 raise "TransferError"
1121 d.addCallback(accepted)
1122 d.addErrback(failed)
1123
1124 # Let the request get pushed to client1
1125 self._loop(10)
1126
1127 # Decline the file
1128 fileReceive = self.client1.fileReceive
1129 fileReceive.reject()
1130
1131 # Let the decline get pushed to client2
1132 self._loop(10)
1133
1134 self.failUnless((self.gotDecline), "Failed to understand a decline, ignored.")
1135
1136
1137 ################
1138 # MSNFTP tests #
1139 ################
1140
1141 class FileTransferTestCase(unittest.TestCase):
1142 """ test FileSend against FileReceive """
1143 skip = "Not implemented"
1144
1145 def setUp(self):
1146 self.input = StringIOWithoutClosing()
1147 self.input.writelines(['a'] * 7000)
1148 self.input.seek(0)
1149 self.output = StringIOWithoutClosing()
1150
1151 def tearDown(self):
1152 self.input = None
1153 self.output = None
1154
1155 def testFileTransfer(self):
1156 auth = 1234
1157 sender = msnft.MSNFTP_FileSend(self.input)
1158 sender.auth = auth
1159 sender.fileSize = 7000
1160 client = msnft.MSNFTP_FileReceive(auth, "foo@bar.com", self.output)
1161 client.fileSize = 7000
1162 loop = LoopbackCon(client, sender)
1163 loop.doSteps(100)
1164 self.failUnless((client.completed and sender.completed), "send failed to complete")
1165 self.failUnless((self.input.getvalue() == self.output.getvalue()), "saved file does not match original")
1166
1167
Python based MSN transport for Jabber