]> code.delx.au - pymsnt/blobdiff - src/tlib/msn/test_msnw.py
File transfer test works! We can send files to MSN Messenger 7. Jabber glue not done...
[pymsnt] / src / tlib / msn / test_msnw.py
index 0dfd652049c9f88cfcc7e28fa918cbfa5f67d4c6..fe078546ab466580bc5ff753b1a1bbea5a099a99 100644 (file)
@@ -20,7 +20,7 @@ import msnw
 
 # Settings
 TIMEOUT = 30.0 # Connection timeout in seconds
-LOGGING = False
+LOGGING = True
 USER1 = "messengertest1@hotmail.com"
 PASS1 = "hellohello"
 USER2 = "messengertest2@hotmail.com"
@@ -105,36 +105,38 @@ class TestsUtil:
                        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()
@@ -160,15 +162,18 @@ class BasicTests(unittest.TestCase, TestsUtil):
 
        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()
@@ -177,5 +182,28 @@ class BasicTests(unittest.TestCase, TestsUtil):
                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)