# Settings
TIMEOUT = 30.0 # Connection timeout in seconds
-LOGGING = False
+LOGGING = True
USER1 = "messengertest1@hotmail.com"
PASS1 = "hellohello"
USER2 = "messengertest2@hotmail.com"
self.user2 = MSNConnection(USER2, PASS2, "user2", self)
self.loop("Logging in user2.", cond="SYNCED")
- def doPurgeContacts(self):
+ def doPurgeContacts(self, both=True):
# Purge both contact lists
clearAccount(self.user1).addCallback(self.cb)
self.loop("Purging user1 contact list.")
- clearAccount(self.user2).addCallback(self.cb)
- self.loop("Purging user2 contact list.")
+ if both:
+ clearAccount(self.user2).addCallback(self.cb)
+ self.loop("Purging user2 contact list.")
- def doAddContacts(self):
+ def doAddContacts(self, both=True):
# Adding users to each other's lists
self.user1.addContact(msnw.FORWARD_LIST, USER2).addCallback(self.cb)
self.loop("Adding user2 to user1's forward list.")
self.user1.addContact(msnw.ALLOW_LIST, USER2).addCallback(self.cb)
self.loop("Adding user2 to user1's allow list.")
- self.user2.addContact(msnw.FORWARD_LIST, USER1).addCallback(self.cb)
- self.loop("Adding user1 to user2's forward list.")
- self.user2.addContact(msnw.ALLOW_LIST, USER1).addCallback(self.cb)
- self.loop("Adding user1 to user2's allow list.")
+ if both:
+ self.user2.addContact(msnw.FORWARD_LIST, USER1).addCallback(self.cb)
+ self.loop("Adding user1 to user2's forward list.")
+ self.user2.addContact(msnw.ALLOW_LIST, USER1).addCallback(self.cb)
+ self.loop("Adding user1 to user2's allow list.")
- # Check the contacts have seen each other
- reactor.iterate(0.1) # One last chance to notice each other
- self.failUnless((self.user1.contactAdded == USER2 and self.user2.contactAdded == USER1), "Contacts can't see each other.")
+ # Check the contacts have seen each other
+ reactor.iterate(0.1) # One last chance to notice each other
+ self.failUnless((self.user1.contactAdded == USER2 and self.user2.contactAdded == USER1), "Contacts can't see each other.")
def cb(self, ignored=None):
self.done = True
- def loop(self, failMsg, cond=None):
+ def loop(self, failMsg, cond=True, timeout=TIMEOUT):
# Loops with a timeout
self.done = False
- self.timeout = reactor.callLater(TIMEOUT, self.failed, "Timeout: " + failMsg)
- while not self.done:
+ self.timeout = reactor.callLater(timeout, self.failed, "Timeout: " + failMsg)
+ while self.done != cond and not self.done:
reactor.iterate(0.1)
try:
self.timeout.cancel()
def testConnect(self):
self.doLogins(both=False)
+# testConnect.skip = "True"
def testPurgeContacts(self):
self.doLogins()
self.doPurgeContacts()
+# testPurgeContacts.skip = "True"
def testAddContacts(self):
self.doLogins()
self.doPurgeContacts()
self.doAddContacts()
+# testAddContacts.skip = "True"
def testMessageExchange(self):
self.doLogins()
self.user1.sendMessage(USER2, "Hi user2")
self.loop("Timeout exchanging message.", cond="GOTMESSAGE")
self.failUnless((self.user2.message == (USER1, "Hi user2")), "Failed to transfer message.")
-
+# testMessageExchange.skip = "True"
+
+ def testFileTransfer(self):
+ data = "Testing 123\r\n" * 5000
+ def accepted((yes,)):
+ if yes:
+ self.fileSend.write(data)
+ self.fileSend.close()
+ else:
+ self.fail("File was not accepted.")
+ def failed():
+ self.fail("Transfer failed in invitation.")
+ def gotFileSend((fileSend, d)):
+ self.fileSend = fileSend
+ d.addCallbacks(accepted, failed)
+ if raw_input("\n\nALERT!!!\n\nPlease 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.\nType ok when you are ready: " % (USER2, USER1)).lower() != "ok":
+ print "TEST SKIPPED!!!"
+ return
+ self.doLogins(both=False)
+ self.doPurgeContacts(both=False)
+ self.doAddContacts(both=False)
+ d = self.user1.sendFile(USER2, "myfile.txt", len(data))
+ d.addCallback(gotFileSend)
+ self.loop("Sending file.", cond="GOTMESSAGE", timeout=60*60)