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

   1 # Copyright 2005-2006 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 # Settings
   9 TIMEOUT = 30.0 # Connection timeout in seconds
  10 LOGGING = True
  11 TWOFILES = False
  12 FTSENDTEST = False
  13 FTRECEIVETEST = False
  14 USER1 = "messengertest1@hotmail.com"
  15 PASS1 = "hellohello"
  16 USER2 = "messengertest2@hotmail.com"
  17 PASS2 = "hellohello"
  18
  19
  20
  21 # Twisted imports
  22 from twisted.internet.defer import Deferred
  23 from twisted.internet import reactor, error
  24 from twisted.trial import unittest
  25 from twisted.python import log
  26
  27 # System imports
  28 import sys, random
  29
  30 # Local imports
  31 import msnw
  32 import msn
  33
  34
  35 if LOGGING:
  36         log.startLogging(sys.stdout)
  37
  38
  39 checkCount = 0 # Uck!
  40 def clearAccount(msncon):
  41         """ Clears the contact list of the given MSNConnection. Returns a
  42         Deferred which fires when the task is complete.
  43         """
  44         d = Deferred()
  45         count = 0
  46         global checkCount
  47         checkCount = 0
  48         def cb(ignored=None):
  49                 global checkCount
  50                 checkCount += 1
  51                 if checkCount == count:
  52                         d.callback(None)
  53
  54         for msnContact in msncon.getContacts().contacts.values():
  55                 for list in [msn.FORWARD_LIST, msn.BLOCK_LIST, msn.ALLOW_LIST, msn.PENDING_LIST]:
  56                         if msnContact.lists & list:
  57                                 msncon.remContact(list, msnContact.userHandle).addCallback(cb)
  58                                 count += 1
  59
  60         if count == 0:
  61                 reactor.callLater(0, d.callback, None)
  62         return d
  63
  64
  65 ####################
  66 # Basic connection #
  67 ####################
  68
  69 class MSNConnection(msnw.MSNConnection):
  70         def __init__(self, username, password, ident, testCase):
  71                 msnw.MSNConnection.__init__(self, username, password, ident)
  72                 self.testCase = testCase
  73                 self.message = None
  74                 self.contactAdded = None
  75
  76         def listSynchronized(self):
  77                 # Now we're fully connected
  78                 self.testCase.done = "SYNCED"
  79
  80         def gotMessage(self, userHandle, text):
  81                 self.testCase.done = "GOTMESSAGE"
  82                 self.message = (userHandle, text)
  83
  84         def contactAddedMe(self, userHandle):
  85                 self.contactAdded = userHandle
  86
  87
  88 class TestsUtil:
  89         def setUp(self):
  90                 self.failure = None
  91                 self.timeout = None
  92                 self.done = False
  93                 self.user1 = None
  94                 self.user2 = None
  95
  96         def tearDown(self):
  97                 if self.user1:
  98                         self.user1.logOut()
  99                         reactor.iterate(0.1)
 100                 if self.user2:
 101                         self.user2.logOut()
 102                         reactor.iterate(0.1)
 103
 104         def doLogins(self, both=True):
 105                 # Connect two accounts
 106                 self.user1 = MSNConnection(USER1, PASS1, "user1", self)
 107                 self.loop("Logging in user1.", cond="SYNCED")
 108                 if both:
 109                         self.user2 = MSNConnection(USER2, PASS2, "user2", self)
 110                         self.loop("Logging in user2.", cond="SYNCED")
 111
 112         def doPurgeContacts(self, both=True):
 113                 # Purge both contact lists
 114                 clearAccount(self.user1).addCallback(self.cb)
 115                 self.loop("Purging user1 contact list.")
 116                 if both:
 117                         clearAccount(self.user2).addCallback(self.cb)
 118                         self.loop("Purging user2 contact list.")
 119
 120         def doAddContacts(self, both=True):
 121                 # Adding users to each other's lists
 122                 self.user1.addContact(msn.FORWARD_LIST, USER2).addCallback(self.cb)
 123                 self.loop("Adding user2 to user1's forward list.")
 124                 self.user1.addContact(msn.ALLOW_LIST, USER2).addCallback(self.cb)
 125                 self.loop("Adding user2 to user1's allow list.")
 126                 if both:
 127                         self.user2.addContact(msn.FORWARD_LIST, USER1).addCallback(self.cb)
 128                         self.loop("Adding user1 to user2's forward list.")
 129                         self.user2.addContact(msn.ALLOW_LIST, USER1).addCallback(self.cb)
 130                         self.loop("Adding user1 to user2's allow list.")
 131
 132                         # Check the contacts have seen each other
 133                         reactor.iterate(0.1) # One last chance to notice each other
 134                         self.failUnless((self.user1.contactAdded == USER2 and self.user2.contactAdded == USER1), "Contacts can't see each other.")
 135
 136         def cb(self, ignored=None):
 137                 self.done = True
 138
 139         def loop(self, failMsg, cond=True, timeout=TIMEOUT):
 140                 # Loops with a timeout
 141                 self.done = False
 142                 self.timeout = reactor.callLater(timeout, self.failed, "Timeout: " + failMsg)
 143                 if cond == True:
 144                         while not self.done:
 145                                 reactor.iterate(0.1)
 146                 else:
 147                         while self.done != cond:
 148                                 reactor.iterate(0.1)
 149                 try:
 150                         self.timeout.cancel()
 151                 except (error.AlreadyCancelled, error.AlreadyCalled):
 152                         pass
 153                 if self.failure:
 154                         self.fail(self.failure)
 155                 if cond:
 156                         self.failUnless((self.done == cond), "Failed: " + failMsg)
 157
 158         def failed(self, why):
 159                 self.failure = why
 160                 self.done = True
 161
 162 # The tests!
 163
 164 class BasicTests(unittest.TestCase, TestsUtil):
 165         def setUp(self):
 166                 TestsUtil.setUp(self)
 167
 168         def tearDown(self):
 169                 TestsUtil.tearDown(self)
 170
 171         def testConnect(self):
 172                 self.doLogins(both=False)
 173         testConnect.skip = FTRECEIVETEST or FTSENDTEST
 174
 175         def testPurgeContacts(self):
 176                 self.doLogins()
 177                 self.doPurgeContacts()
 178         testPurgeContacts.skip = FTRECEIVETEST or FTSENDTEST
 179
 180         def testAddContacts(self):
 181                 self.doLogins()
 182                 self.doPurgeContacts()
 183                 self.doAddContacts()
 184         testAddContacts.skip = FTRECEIVETEST or FTSENDTEST
 185
 186         def testMessageExchange(self):
 187                 self.doLogins()
 188                 self.doPurgeContacts()
 189                 self.doAddContacts()
 190                 self.user1.sendMessage(USER2, "Hi user2")
 191                 self.loop("Timeout exchanging message.", cond="GOTMESSAGE")
 192                 self.failUnless((self.user2.message == (USER1, "Hi user2")), "Failed to transfer message.")
 193         testMessageExchange.skip = FTRECEIVETEST or FTSENDTEST
 194
 195         def testFileSend(self):
 196                 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":
 197                         raise unittest.SkipTest("User didn't type 'ok'")
 198
 199                 data = "Testing 123\r\n" * 5000
 200                 def accepted((yes,)):
 201                         if yes:
 202                                 self.fileSend.write(data)
 203                                 self.fileSend.close()
 204                         else:
 205                                 self.fail("File was not accepted.")
 206                 def failed():
 207                         self.fail("Transfer failed in invitation.")
 208                 def gotFileSend((fileSend, d)):
 209                         self.fileSend = fileSend
 210                         d.addCallbacks(accepted, failed)
 211
 212                 self.doLogins(both=False)
 213                 self.doPurgeContacts(both=False)
 214                 self.doAddContacts(both=False)
 215                 d = self.user1.sendFile(USER2, "myfile.txt", len(data))
 216                 d.addCallback(gotFileSend)
 217                 self.loop("Sending file.", cond="GOTMESSAGE", timeout=60*60)
 218                 global TWOFILES
 219                 if TWOFILES:
 220                         d = self.user1.sendFile(USER2, "myfile2.txt", len(data))
 221                         d.addCallback(gotFileSend)
 222                         self.loop("Sending file.", cond="GOTMESSAGE", timeout=60*60)
 223         testFileSend.skip = not FTSENDTEST
 224
 225         def testFileReceive(self):
 226                 if raw_input("\n\nALERT!!!\n\nPlease connect to account %s and send a file transfer to %s.\nType ok when you are ready: " % (USER2, USER1)).lower() != "ok":
 227                         raise unittest.SkipTest("User didn't type 'ok'")
 228
 229                 def fileFinished(data):
 230                         #filename = "/tmp/msn" + str(random.randint(1000, 9999)) + ".dat"
 231                         filename = "/tmp/MSNFILE_" + self.fileReceive.filename
 232                         f = open(filename, "w")
 233                         f.write(data)
 234                         f.close()
 235                         print "Got file!", filename
 236                         # Terminate the loop in a little, let them send the BYE before
 237                         # we drop the connection
 238                         def wait():
 239                                 self.done = "GOTFILE"
 240                         reactor.callLater(5, wait)
 241
 242                 def gotFileReceive(fileReceive):
 243                         buffer = msn.StringBuffer(fileFinished)
 244                         self.fileReceive = fileReceive
 245                         self.fileReceive.accept(buffer)
 246
 247                 self.doLogins(both=False)
 248                 self.user1.gotFileReceive = gotFileReceive
 249                 self.doPurgeContacts(both=False)
 250                 self.doAddContacts(both=False)
 251                 self.loop("Receiving file.", cond="GOTFILE", timeout=60*60)
 252                 global TWOFILES
 253                 if TWOFILES:
 254                         self.loop("Receiving file.", cond="GOTFILE", timeout=60*60)
 255         testFileReceive.skip = not FTRECEIVETEST
 256