import calendar
import threading
import re
+import select
import socket
from test.support import verbose, run_with_tz, run_with_locale, cpython_only
return Handler
+def _read_literal(handler, marker):
+ # Read one literal, a raw octet sequence, by its count from the marker
+ # ('{N}', or '(~{N}' in UTF8 mode).
+ size = int(re.search(r'\{(\d+)\}', marker).group(1))
+ # The client must wait for the continuation, so nothing should be readable.
+ if select.select([handler.connection], [], [], 0)[0]:
+ raise AssertionError('client sent the literal before the '
+ 'continuation request')
+ handler._send_textline('+')
+ return handler.rfile.read(size)
+
+
class TestImaplib(unittest.TestCase):
def test_Internaldate2tuple(self):
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
def cmd_APPEND(self, tag, args):
- self._send_textline('+')
self.server.response = args
- literal = yield
- self.server.response.append(literal)
+ self.server.response.append(_read_literal(self, args[-1]))
literal = yield
self.server.response.append(literal)
self._send_tagged(tag, 'OK', 'okay')
self.assertEqual(data[0], b'LOGIN completed')
self.assertEqual(client.state, 'AUTH')
+ def test_append_line_endings(self):
+ # append() normalizes bare CR and LF in the message to CRLF.
+ class AppendHandler(SimpleIMAPHandler):
+ def cmd_APPEND(self, tag, args):
+ self.server.response = _read_literal(self, args[-1])
+ yield # read the trailer line
+ self._send_tagged(tag, 'OK', 'APPEND completed')
+ client, server = self._setup(AppendHandler)
+ client.login('user', 'pass')
+ message = b'a\rb\nc\r\nd'
+ client.append('INBOX', None, None, message)
+ self.assertEqual(server.response, b'a\r\nb\r\nc\r\nd')
+
def test_login_capabilities(self):
# A server may advertise new capabilities after login (as an
# untagged CAPABILITY response); imaplib must refresh its cached
class UTF8AppendServer(self.UTF8Server):
def cmd_APPEND(self, tag, args):
- self._send_textline('+')
self.server.response = args
- literal = yield
- self.server.response.append(literal)
+ self.server.response.append(_read_literal(self, args[-1]))
literal = yield
self.server.response.append(literal)
self._send_tagged(tag, 'OK', 'okay')