Changeset 218:ae8dacf52b3d

Timestamp:

16/04/11 17:21:28 (13 months ago)

Author:

Menno Smits <menno@…>

Branch:

default

Message:

Added livetest for IDLE support and reworked how the IDLE callback function is called

Files:

• imapclient/imapclient.py (modified) (6 diffs)
• livetest.py (modified) (2 diffs)

imapclient/imapclient.py

@@ -10 +10 @@
 try:
     import imaplib2 as imaplib
-    imaplib2 = True
+    using_imaplib2 = True
 except ImportError:
-    imaplib2 = False
+    using_imaplib2 = False
     import imaplib
     
@@ -27 +27 @@
 
 __all__ = ['IMAPClient', 'DELETED', 'SEEN', 'ANSWERED', 'FLAGGED', 'DRAFT',
-    'RECENT']
+    'RECENT', 'using_imaplib2']
 
 from response_parser import parse_response, parse_fetch_response
@@ -295 +295 @@
         typ, data = self._imap.select(self._encode_folder_name(folder), readonly)
         self._checkok('select', typ, data)
-        if imaplib2:
+        if using_imaplib2:
             untagged_responses = self._response_to_dict(self._imap.untagged_responses)
         else:
@@ -326 +326 @@
         return out
 
-    def idle(self, timeout=None, **kwargs):
+    def idle(self, timeout=None, callback=None, **kwargs):
         """Put server into IDLE mode.
 
-        IDLE mode will end when the server notifies of a change, the sever
+        IDLE mode will end when the server notifies of a change, the server
         reaches the timeout, or another IMAP4 command is scheduled.
         
@@ -337 +337 @@
         @param cb_arg: An optional argument that will be passed to the callback function.
         @return: Server response. (None if callback is specified).
-        """
-        if imaplib2:
-            if 'callback' in kwargs:
-                callback = kwargs.pop('callback')
-                def _wrapped_callback(resp):
+
+        #XXX update documentation
+        #XXX what about finding out what the IDLE data was?
+        """
+        if not using_imaplib2:
+            raise self.Error('The imaplib2 module is required to use IDLE')
+
+        if callback:
+            cb_arg_was_given = 'cb_arg' in kwargs
+            def _wrapped_cb(resp):
+                if resp[0]:
+                    success = True
                     typ, data = resp[0]
-                    self._checkok('idle', typ, data)
-                    # if there's a cb_arg, pass it along
+                    response = (typ, data[0])
+                else:
+                    success = False
+                    response = resp[2]
+                if cb_arg_was_given:
                     cb_arg = resp[1]
-                    if cb_arg:
-                        callback(data[0], cb_arg)
-                    else:
-                        callback(data[0])
-                self._imap.idle(timeout, callback=_wrapped_callback, **kwargs)
-            else:
-                typ, data = self._imap.idle(timeout)
-                self._checkok('idle', typ, data)
-                return data[0]
-        else:
-            raise self.Error('The imaplib2 module is required to use IDLE')
+                    callback(success, response, cb_arg)
+                else:
+                    callback(success, response)
+            self._imap.idle(timeout, callback=_wrapped_cb, cb_arg=kwargs.get('cb_arg'))
+        else:
+            typ, data = self._imap.idle(timeout)
+            self._checkok('idle', typ, data)
+            return data[0]
                 
     def folder_status(self, folder, what=None):
@@ -585 +592 @@
         else:
             tag = self._imap._command('FETCH', msg_list, parts_list, modifiers_list)
-        if imaplib2:
+        if using_imaplib2:
             typ, data = self._imap._command_complete(tag, 'FETCH')
         else:

livetest.py

@@ -9 +9 @@
 import os
 import sys
+import time
+import threading
 from datetime import datetime
 from ConfigParser import SafeConfigParser, NoOptionError
@@ -449 +451 @@
                         e = lower_if_str(e)
                         self.assertEqual(a, e)
+        
+        def test_idle(self):
+            if not imapclient.using_imaplib2:
+                return self.skipTest("imaplib2 is not installed")
+            
+            idle_event = threading.Event()
+            cb_data = {}
+            def cb(success, response, cb_arg):
+                idle_event.set()
+                cb_data['success'] = success
+                cb_data['idle_response'] = response
+                cb_data['cb_arg'] = cb_arg
+
+            # Start main connection idling
+            self.client.select_folder('INBOX')
+            self.client.idle(timeout=20, callback=cb, cb_arg='foo')
+
+            # Start a new connection and upload a new message
+            client2 = create_client_from_config(conf)
+            client2.select_folder('INBOX')
+            client2.append('INBOX', SIMPLE_MESSAGE)
+            client2.logout()
+
+            idle_event.wait(15)    # Wait for IDLE callback to trigger
+
+            self.assertTrue(idle_event.is_set())
+            self.assertTrue(cb_data['success'])
+            idle_status, idle_text = cb_data['idle_response']
+            self.assertEqual(idle_status, 'OK')
+            # Can't be too specific as different servers return different text
+            self.assertIn('idle', idle_text.lower())     
+            self.assertEqual(cb_data['cb_arg'], 'foo')
+            
     return LiveTest