code.delx.au - pymsnt/blob - src/tlib/msn/test_msnw.py

   1 # Copyright 2005 James Bunton <james@delx.cjb.net>
   2 # Licensed for distribution under the GPL version 2, check COPYING for details
   3
   4 """
   5 Test cases for msnw (MSN Wrapper)
   6 """
   7
   8 # Twisted imports
   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
  13
  14 # System imports
  15 import sys
  16
  17 # Local imports
  18 import msnw
  19
  20
  21 # Settings
  22 TIMEOUT = 30.0 # Connection timeout in seconds
  23 LOGGING = True
  24 USER1 = "messengertest1@hotmail.com"
  25 PASS1 = "hellohello"
  26 USER2 = "messengertest2@hotmail.com"
  27 PASS2 = "hellohello"
  28
  29
  30
  31 if LOGGING:
  32         log.startLogging(sys.stdout)
  33
  34
  35 checkCount = 0 # Uck!
  36 def clearAccount(msncon):
  37         """ Clears the contact list of the given MSNConnection. Returns a
  38         Deferred which fires when the task is complete.
  39         """
  40         d = Deferred()
  41         count = 0
  42         global checkCount
  43         checkCount = 0
  44         def cb(ignored=None):
  45                 global checkCount
  46                 checkCount += 1
  47                 if checkCount == count:
  48                         d.callback(None)
  49
  50         for msnContact in msncon.getContacts().contacts.values():
  51                 for list in [msnw.FORWARD_LIST, msnw.BLOCK_LIST, msnw.ALLOW_LIST, msnw.PENDING_LIST]:
  52                         if msnContact.lists & list:
  53                                 msncon.remContact(list, msnContact.userHandle).addCallback(cb)
  54                                 count += 1
  55
  56         if count == 0:
  57                 reactor.callLater(0, d.callback, None)
  58         return d
  59
  60
  61 ####################
  62 # Basic connection #
  63 ####################
  64
  65 class MSNConnection(msnw.MSNConnection):
  66         def __init__(self, username, password, ident, testCase):
  67                 msnw.MSNConnection.__init__(self, username, password, ident)
  68                 self.testCase = testCase
  69                 self.message = None
  70                 self.contactAdded = None
  71
  72         def listSynchronized(self):
  73                 # Now we're fully connected
  74                 self.testCase.done = "SYNCED"
  75
  76         def gotMessage(self, userHandle, text):
  77                 self.testCase.done = "GOTMESSAGE"
  78                 self.message = (userHandle, text)
  79
  80         def contactAddedMe(self, userHandle):
  81                 self.contactAdded = userHandle
  82
  83
  84 class TestsUtil:
  85         def setUp(self):
  86                 self.failure = None
  87                 self.timeout = None
  88                 self.done = False
  89                 self.user1 = None
  90                 self.user2 = None
  91
  92         def tearDown(self):
  93                 if self.user1:
  94                         self.user1.logOut()
  95                         reactor.iterate(0.1)
  96                 if self.user2:
  97                         self.user2.logOut()
  98                         reactor.iterate(0.1)
  99
 100         def doLogins(self, both=True):
 101                 # Connect two accounts
 102                 self.user1 = MSNConnection(USER1, PASS1, "user1", self)
 103                 self.loop("Logging in user1.", cond="SYNCED")
 104                 if both:
 105                         self.user2 = MSNConnection(USER2, PASS2, "user2", self)
 106                         self.loop("Logging in user2.", cond="SYNCED")
 107
 108         def doPurgeContacts(self, both=True):
 109                 # Purge both contact lists
 110                 clearAccount(self.user1).addCallback(self.cb)
 111                 self.loop("Purging user1 contact list.")
 112                 if both:
 113                         clearAccount(self.user2).addCallback(self.cb)
 114                         self.loop("Purging user2 contact list.")
 115
 116         def doAddContacts(self, both=True):
 117                 # Adding users to each other's lists
 118                 self.user1.addContact(msnw.FORWARD_LIST, USER2).addCallback(self.cb)
 119                 self.loop("Adding user2 to user1's forward list.")
 120                 self.user1.addContact(msnw.ALLOW_LIST, USER2).addCallback(self.cb)
 121                 self.loop("Adding user2 to user1's allow list.")
 122                 if both:
 123                         self.user2.addContact(msnw.FORWARD_LIST, USER1).addCallback(self.cb)
 124                         self.loop("Adding user1 to user2's forward list.")
 125                         self.user2.addContact(msnw.ALLOW_LIST, USER1).addCallback(self.cb)
 126                         self.loop("Adding user1 to user2's allow list.")
 127
 128                         # Check the contacts have seen each other
 129                         reactor.iterate(0.1) # One last chance to notice each other
 130                         self.failUnless((self.user1.contactAdded == USER2 and self.user2.contactAdded == USER1), "Contacts can't see each other.")
 131
 132         def cb(self, ignored=None):
 133                 self.done = True
 134
 135         def loop(self, failMsg, cond=True, timeout=TIMEOUT):
 136                 # Loops with a timeout
 137                 self.done = False
 138                 self.timeout = reactor.callLater(timeout, self.failed, "Timeout: " + failMsg)
 139                 while self.done != cond and not self.done:
 140                         reactor.iterate(0.1)
 141                 try:
 142                         self.timeout.cancel()
 143                 except (error.AlreadyCancelled, error.AlreadyCalled):
 144                         pass
 145                 if self.failure:
 146                         self.fail(self.failure)
 147                 if cond:
 148                         self.failUnless((self.done == cond), "Failed: " + failMsg)
 149
 150         def failed(self, why):
 151                 self.failure = why
 152                 self.done = True
 153
 154 # The tests!
 155
 156 class BasicTests(unittest.TestCase, TestsUtil):
 157         def setUp(self):
 158                 TestsUtil.setUp(self)
 159
 160         def tearDown(self):
 161                 TestsUtil.tearDown(self)
 162
 163         def testConnect(self):
 164                 self.doLogins(both=False)
 165 #       testConnect.skip = "True"
 166
 167         def testPurgeContacts(self):
 168                 self.doLogins()
 169                 self.doPurgeContacts()
 170 #       testPurgeContacts.skip = "True"
 171
 172         def testAddContacts(self):
 173                 self.doLogins()
 174                 self.doPurgeContacts()
 175                 self.doAddContacts()
 176 #       testAddContacts.skip = "True"
 177
 178         def testMessageExchange(self):
 179                 self.doLogins()
 180                 self.doPurgeContacts()
 181                 self.doAddContacts()
 182                 self.user1.sendMessage(USER2, "Hi user2")
 183                 self.loop("Timeout exchanging message.", cond="GOTMESSAGE")
 184                 self.failUnless((self.user2.message == (USER1, "Hi user2")), "Failed to transfer message.")
 185 #       testMessageExchange.skip = "True"
 186
 187         def testFileTransfer(self):
 188                 data = "Testing 123\r\n" * 5000
 189                 def accepted((yes,)):
 190                         if yes:
 191                                 self.fileSend.write(data)
 192                                 self.fileSend.close()
 193                         else:
 194                                 self.fail("File was not accepted.")
 195                 def failed():
 196                         self.fail("Transfer failed in invitation.")
 197                 def gotFileSend((fileSend, d)):
 198                         self.fileSend = fileSend
 199                         d.addCallbacks(accepted, failed)
 200                 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":
 201                         print "TEST SKIPPED!!!"
 202                         return
 203                 self.doLogins(both=False)
 204                 self.doPurgeContacts(both=False)
 205                 self.doAddContacts(both=False)
 206                 d = self.user1.sendFile(USER2, "myfile.txt", len(data))
 207                 d.addCallback(gotFileSend)
 208                 self.loop("Sending file.", cond="GOTMESSAGE", timeout=60*60)
 209