diff --git a/Lib/test/test_imaplib.py b/Lib/test/test_imaplib.py --- a/Lib/test/test_imaplib.py +++ b/Lib/test/test_imaplib.py @@ -172,6 +172,119 @@ self._send_tagged(tag, 'OK', 'LOGIN completed') +class NewIMAPTests(unittest.TestCase): + _client = None + + def _setup(self, imap_server, connect=True): + """ + Setups imap_server (which should inherit SimpleIMAPHandler) for tests. + Returns (client, server). + """ + class TestTCPServer(socketserver.TCPServer): + def handle_error(self, request, client_address): + """ + End request and raise the error if one occurs. + """ + self.close_request(request) + self.server_close() + raise + + self.addCleanup(self._cleanup) + self._server = TestTCPServer((support.HOST, 0), imap_server) + self._thread = threading.Thread( + name=self._server.__class__.__name__, + target=self._server.serve_forever, + # Short poll interval to make the test finish quickly. + # Time between requests is short enough that we won't wake + # up spuriously too many times. + kwargs={'poll_interval': 0.01}) + self._thread.daemon = True # In case this function raises. + self._thread.start() + + if connect: + self._client = imaplib.IMAP4(*self._server.server_address) + + return self._client, self._server + + def _cleanup(self): + """ + Cleans up the test server. This method should not be called manually, + it's being add to cleanup queue in the _setup method already. + """ + # if logout was called already we'd raise an exception trying to + # shutdown the client once again + if self._client is not None and self._client.state != 'LOGOUT': + self._client.shutdown() + # cleanup the server + self._server.shutdown() + self._server.server_close() + self._thread.join() + + def test_issue5949(self): + class EOFHandler(socketserver.StreamRequestHandler): + def handle(self): + # EOF without sending a complete welcome message. + self.wfile.write(b'* OK') + _, server = self._setup(EOFHandler, connect=False) + self.assertRaises(imaplib.IMAP4.abort, imaplib.IMAP4, *server.server_address) + + def test_line_termination(self): + class BadNewlineHandler(SimpleIMAPHandler): + def cmd_CAPABILITY(self, tag, args): + self._send(b'* CAPABILITY IMAP4rev1 AUTH\n') + self._send_tagged(tag, 'OK', 'CAPABILITY completed') + _, server = self._setup(BadNewlineHandler, connect=False) + self.assertRaises(imaplib.IMAP4.abort, imaplib.IMAP4, *server.server_address) + + def test_enable_raises_error_if_not_AUTH(self): + class EnableHandler(SimpleIMAPHandler): + capabilities = 'AUTH ENABLE UTF8=ACCEPT' + client, _ = self._setup(EnableHandler) + self.assertFalse(client.utf8_enabled) + self.assertRaisesRegex(imaplib.IMAP4.error, 'command ENABLE illegal in state NONAUTH', client.enable, 'foo') + self.assertFalse(client.utf8_enabled) + + def test_enable_raises_error_if_no_capability(self): + client, _ = self._setup(SimpleIMAPHandler) + self.assertRaisesRegex(imaplib.IMAP4.error, 'Server does not support ENABLE', client.enable, 'foo') + + def test_enable_UTF8_raises_error_if_not_supported(self): + client, _ = self._setup(SimpleIMAPHandler) + typ, data = client.login('user', 'pass') + self.assertEqual(typ, 'OK') + self.assertRaisesRegex(imaplib.IMAP4.error, 'Server does not support ENABLE', client.enable, 'UTF8=ACCEPT') + + # command tests + + def test_login(self): + client, _ = self._setup(SimpleIMAPHandler) + typ, data = client.login('user', 'pass') + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'LOGIN completed') + self.assertEqual(client.state, 'AUTH') + + def test_logout(self): + client, _ = self._setup(SimpleIMAPHandler) + typ, data = client.login('user', 'pass') + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'LOGIN completed') + typ, data = client.logout() + self.assertEqual(typ, 'BYE') + self.assertEqual(data[0], b'IMAP4ref1 Server logging out') + self.assertEqual(client.state, 'LOGOUT') + + def test_lsub(self): + class LsubCmd(SimpleIMAPHandler): + def cmd_LSUB(self, tag, args): + self._send_textline('* LSUB () "." directoryA') + return self._send_tagged(tag, 'OK', 'LSUB completed') + client, _ = self._setup(LsubCmd) + client.login('user', 'pass') + typ, data = client.lsub() + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'() "." directoryA') + + class ThreadedNetworkedTests(unittest.TestCase): server_class = socketserver.TCPServer imap_class = imaplib.IMAP4