# Copyright (c) 2014, Menno Smits # Released subject to the New BSD License # Please see http://en.wikipedia.org/wiki/BSD_licenses from __future__ import unicode_literals import itertools import socket import sys from datetime import datetime import six from mock import patch, sentinel, Mock from imapclient.fixed_offset import FixedOffset from .testable_imapclient import TestableIMAPClient as IMAPClient from .imapclient_test import IMAPClientTest class TestListFolders(IMAPClientTest): def test_list_folders(self): self.client._imap._simple_command.return_value = ('OK', [b'something']) self.client._imap._untagged_response.return_value = ('LIST', sentinel.folder_data) self.client._proc_folder_list = Mock(return_value=sentinel.folder_list) folders = self.client.list_folders('foo', 'bar') self.client._imap._simple_command.assert_called_once_with( 'LIST', b'"foo"', b'"bar"') self.assertEqual(self.client._proc_folder_list.call_args, ((sentinel.folder_data,), {})) self.assertTrue(folders is sentinel.folder_list) def test_list_sub_folders(self): self.client._imap._simple_command.return_value = ('OK', [b'something']) self.client._imap._untagged_response.return_value = ('LSUB', sentinel.folder_data) self.client._proc_folder_list = Mock(return_value=sentinel.folder_list) folders = self.client.list_sub_folders('foo', 'bar') self.client._imap._simple_command.assert_called_once_with( 'LSUB', b'"foo"', b'"bar"') self.assertEqual(self.client._proc_folder_list.call_args, ((sentinel.folder_data,), {})) self.assertTrue(folders is sentinel.folder_list) def test_list_folders_NO(self): self.client._imap._simple_command.return_value = ('NO', [b'badness']) self.assertRaises(IMAPClient.Error, self.client.list_folders) def test_list_sub_folders_NO(self): self.client._imap._simple_command.return_value = ('NO', [b'badness']) self.assertRaises(IMAPClient.Error, self.client.list_folders) def test_utf7_decoding(self): self.client._imap._simple_command.return_value = ('OK', [b'something']) self.client._imap._untagged_response.return_value = ( 'LIST', [ b'(\\HasNoChildren) "/" "A"', b'(\\HasNoChildren) "/" "Hello&AP8-world"', ]) folders = self.client.list_folders('foo', 'bar') self.client._imap._simple_command.assert_called_once_with('LIST', b'"foo"', b'"bar"') self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', 'A'), ((b'\\HasNoChildren',), b'/', 'Hello\xffworld')]) def test_folder_encode_off(self): self.client.folder_encode = False self.client._imap._simple_command.return_value = ('OK', [b'something']) self.client._imap._untagged_response.return_value = ( 'LIST', [ b'(\\HasNoChildren) "/" "A"', b'(\\HasNoChildren) "/" "Hello&AP8-world"', ]) folders = self.client.list_folders('foo', 'bar') self.client._imap._simple_command.assert_called_once_with('LIST', '"foo"', '"bar"') self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', b'A'), ((b'\\HasNoChildren',), b'/', b'Hello&AP8-world')]) def test_simple(self): folders = self.client._proc_folder_list([b'(\\HasNoChildren) "/" "A"', b'(\\HasNoChildren) "/" "Foo Bar"', ]) self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', 'A',), ((b'\\HasNoChildren',), b'/', 'Foo Bar')]) def test_without_quotes(self): folders = self.client._proc_folder_list([b'(\\HasNoChildren) "/" A', b'(\\HasNoChildren) "/" B', b'(\\HasNoChildren) "/" C', ]) self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', 'A'), ((b'\\HasNoChildren',), b'/', 'B'), ((b'\\HasNoChildren',), b'/', 'C')]) def test_unquoted_numeric_folder_name(self): # Some IMAP implementations do this folders = self.client._proc_folder_list([b'(\\HasNoChildren) "/" 123']) self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', '123')]) def test_unqouted_numeric_folder_name_parsed_as_long(self): # big enough numeric values might get parsed as longs folder_name = str(sys.maxsize + 1) folders = self.client._proc_folder_list( [b'(\\HasNoChildren) "/" ' + folder_name.encode('ascii')]) self.assertEqual(folders, [((b'\\HasNoChildren', ), b'/', folder_name)]) def test_mixed(self): folders = self.client._proc_folder_list([b'(\\HasNoChildren) "/" Alpha', b'(\\HasNoChildren) "/" "Foo Bar"', b'(\\HasNoChildren) "/" C', ]) self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', 'Alpha'), ((b'\\HasNoChildren',), b'/', 'Foo Bar'), ((b'\\HasNoChildren',), b'/', 'C')]) def test_funky_characters(self): folders = self.client._proc_folder_list([(b'(\\NoInferiors \\UnMarked) "/" {5}', 'bang\xff'), b'', b'(\\HasNoChildren \\UnMarked) "/" "INBOX"']) self.assertEqual(folders, [((b'\\NoInferiors', b'\\UnMarked'), b"/", 'bang\xff'), ((b'\\HasNoChildren', b'\\UnMarked'), b"/", 'INBOX')]) def test_quoted_specials(self): folders = self.client._proc_folder_list([br'(\HasNoChildren) "/" "Test \"Folder\""', br'(\HasNoChildren) "/" "Left\"Right"', br'(\HasNoChildren) "/" "Left\\Right"', br'(\HasNoChildren) "/" "\"Left Right\""', br'(\HasNoChildren) "/" "\"Left\\Right\""', ]) self.assertEqual(folders, [((b'\\HasNoChildren',), b'/', 'Test "Folder"'), ((b'\\HasNoChildren',), b'/', 'Left\"Right'), ((b'\\HasNoChildren',), b'/', r'Left\Right'), ((b'\\HasNoChildren',), b'/', r'"Left Right"'), ((b'\\HasNoChildren',), b'/', r'"Left\Right"'), ]) def test_empty_response(self): self.assertEqual(self.client._proc_folder_list([None]), []) def test_blanks(self): folders = self.client._proc_folder_list(['', None, br'(\HasNoChildren) "/" "last"']) self.assertEqual(folders, [((br'\HasNoChildren',), b'/', 'last')]) class TestSelectFolder(IMAPClientTest): def test_normal(self): self.client._command_and_check = Mock() self.client._imap.untagged_responses = { b'exists': [b'3'], b'FLAGS': [br"(\Flagged \Deleted abc [foo]/bar def)"], b'HIGHESTMODSEQ': [b'127110'], b'OK': [br"[PERMANENTFLAGS (\Flagged \Deleted abc [foo]/bar def \*)] Flags permitted.", b'[UIDVALIDITY 631062293] UIDs valid.', b'[UIDNEXT 1281] Predicted next UID.', b'[HIGHESTMODSEQ 127110]'], b'PERMANENTFLAGS': [br'(\Flagged \Deleted abc [foo'], b'READ-WRITE': [b''], b'RECENT': [b'0'], b'UIDNEXT': [b'1281'], b'UIDVALIDITY': [b'631062293'], b'OTHER': [b'blah'] } result = self.client.select_folder(b'folder_name', sentinel.readonly) self.client._command_and_check.assert_called_once_with('select', b'"folder_name"', sentinel.readonly) self.maxDiff = 99999 self.assertEqual(result, { b'EXISTS': 3, b'RECENT': 0, b'UIDNEXT': 1281, b'UIDVALIDITY': 631062293, b'HIGHESTMODSEQ': 127110, b'FLAGS': (br'\Flagged', br'\Deleted', b'abc', b'[foo]/bar', b'def'), b'PERMANENTFLAGS': (br'\Flagged', br'\Deleted', b'abc', b'[foo]/bar', b'def', br'\*'), b'READ-WRITE': True, b'OTHER': [b'blah'] }) class TestAppend(IMAPClientTest): def test_without_msg_time(self): self.client._imap.append.return_value = ('OK', [b'Good']) msg = 'hi' self.client.append('foobar', msg, ['FLAG', 'WAVE'], None) self.client._imap.append.assert_called_with( b'"foobar"', '(FLAG WAVE)', None, b'hi') @patch('imapclient.imapclient.datetime_to_INTERNALDATE') def test_with_msg_time(self, datetime_to_INTERNALDATE): datetime_to_INTERNALDATE.return_value = 'somedate' self.client._imap.append.return_value = ('OK', [b'Good']) msg = b'bye' self.client.append('foobar', msg, ['FLAG', 'WAVE'], datetime(2009, 4, 5, 11, 0, 5, 0, FixedOffset(2 * 60))) self.assertTrue(datetime_to_INTERNALDATE.called) self.client._imap.append.assert_called_with( b'"foobar"', '(FLAG WAVE)', '"somedate"', msg) class TestAclMethods(IMAPClientTest): def test_getacl(self): self.client._imap.getacl.return_value = ('OK', [b'INBOX Fred rwipslda Sally rwip']) acl = self.client.getacl('INBOX') self.assertSequenceEqual(acl, [(b'Fred', b'rwipslda'), (b'Sally', b'rwip')]) def test_setacl(self): self.client._imap.setacl.return_value = ('OK', [b"SETACL done"]) response = self.client.setacl('folder', sentinel.who, sentinel.what) self.client._imap.setacl.assert_called_with(b'"folder"', sentinel.who, sentinel.what) self.assertEqual(response, b"SETACL done") class TestIdleAndNoop(IMAPClientTest): def assert_sock_calls(self, sock): self.assertListEqual(sock.method_calls, [ ('settimeout', (None,), {}), ('setblocking', (0,), {}), ('setblocking', (1,), {}), ]) def test_idle(self): self.client._imap._command.return_value = sentinel.tag self.client._imap._get_response.return_value = None self.client.idle() self.client._imap._command.assert_called_with('IDLE') self.assertEqual(self.client._idle_tag, sentinel.tag) @patch('imapclient.imapclient.select.select') def test_idle_check_blocking(self, mock_select): mock_sock = Mock() self.client._imap.sock = self.client._imap.sslobj = mock_sock mock_select.return_value = ([True], [], []) counter = itertools.count() def fake_get_line(): count = six.next(counter) if count == 0: return b'* 1 EXISTS' elif count == 1: return b'* 0 EXPUNGE' else: raise socket.timeout self.client._imap._get_line = fake_get_line responses = self.client.idle_check() mock_select.assert_called_once_with([mock_sock], [], [], None) self.assert_sock_calls(mock_sock) self.assertListEqual([(1, b'EXISTS'), (0, b'EXPUNGE')], responses) @patch('imapclient.imapclient.select.select') def test_idle_check_timeout(self, mock_select): mock_sock = Mock() self.client._imap.sock = self.client._imap.sslobj = mock_sock mock_select.return_value = ([], [], []) responses = self.client.idle_check(timeout=0.5) mock_select.assert_called_once_with([mock_sock], [], [], 0.5) self.assert_sock_calls(mock_sock) self.assertListEqual([], responses) @patch('imapclient.imapclient.select.select') def test_idle_check_with_data(self, mock_select): mock_sock = Mock() self.client._imap.sock = self.client._imap.sslobj = mock_sock mock_select.return_value = ([True], [], []) counter = itertools.count() def fake_get_line(): count = six.next(counter) if count == 0: return b'* 99 EXISTS' else: raise socket.timeout self.client._imap._get_line = fake_get_line responses = self.client.idle_check() mock_select.assert_called_once_with([mock_sock], [], [], None) self.assert_sock_calls(mock_sock) self.assertListEqual([(99, b'EXISTS')], responses) def test_idle_done(self): self.client._idle_tag = sentinel.tag mockSend = Mock() self.client._imap.send = mockSend mockConsume = Mock(return_value=sentinel.out) self.client._consume_until_tagged_response = mockConsume result = self.client.idle_done() mockSend.assert_called_with(b'DONE\r\n') mockConsume.assert_called_with(sentinel.tag, 'IDLE') self.assertEqual(result, sentinel.out) def test_noop(self): mockCommand = Mock(return_value=sentinel.tag) self.client._imap._command = mockCommand mockConsume = Mock(return_value=sentinel.out) self.client._consume_until_tagged_response = mockConsume result = self.client.noop() mockCommand.assert_called_with('NOOP') mockConsume.assert_called_with(sentinel.tag, 'NOOP') self.assertEqual(result, sentinel.out) def test_consume_until_tagged_response(self): client = self.client client._imap.tagged_commands = {sentinel.tag: None} counter = itertools.count() def fake_get_response(): count = six.next(counter) if count == 0: return b'* 99 EXISTS' client._imap.tagged_commands[sentinel.tag] = ('OK', [b'Idle done']) client._imap._get_response = fake_get_response text, responses = client._consume_until_tagged_response(sentinel.tag, b'IDLE') self.assertEqual(client._imap.tagged_commands, {}) self.assertEqual(text, b'Idle done') self.assertListEqual([(99, b'EXISTS')], responses) class TestDebugLogging(IMAPClientTest): def test_default_is_stderr(self): self.assertIs(self.client.log_file, sys.stderr) def test_IMAP_is_patched(self): log = six.StringIO() self.client.log_file = log self.client._log('one') self.client._imap._mesg('two') output = log.getvalue() self.assertIn('one', output) self.assertIn('two', output) class TestTimeNormalisation(IMAPClientTest): def test_default(self): self.assertTrue(self.client.normalise_times) @patch('imapclient.imapclient.parse_fetch_response') def test_pass_through(self, parse_fetch_response): self.client._imap._command_complete.return_value = ('OK', sentinel.data) self.client._imap._untagged_response.return_value = ('OK', sentinel.fetch_data) self.client.use_uid = sentinel.use_uid def check(expected): self.client.fetch(22, ['SOMETHING']) parse_fetch_response.assert_called_with(sentinel.fetch_data, expected, sentinel.use_uid) self.client.normalise_times = True check(True) self.client.normalise_times = False check(False) class TestNamespace(IMAPClientTest): def set_return(self, value): self.client._imap.namespace.return_value = ('OK', [value]) def test_simple(self): self.set_return(b'(("FOO." "/")) NIL NIL') self.assertEqual(self.client.namespace(), ((('FOO.', '/'),), None, None)) def test_folder_decoding(self): self.set_return(b'(("&AP8-." "/")) NIL NIL') self.assertEqual(self.client.namespace(), ((('\xff.', '/'),), None, None)) def test_without_folder_decoding(self): self.set_return(b'(("&AP8-." "/")) NIL NIL') self.client.folder_encode = False self.assertEqual(self.client.namespace(), (((b'&AP8-.', '/'),), None, None)) def test_other_only(self): self.set_return(b'NIL NIL (("" "."))') self.assertEqual(self.client.namespace(), (None, None, (("", "."),))) def test_complex(self): self.set_return(b'(("" "/")) ' b'(("~" "/")) ' b'(("#shared/" "/") ("#public/" "/")("#ftp/" "/")("#news." "."))') self.assertEqual(self.client.namespace(), ( (("", "/"),), (("~", "/"),), (("#shared/", "/"), ("#public/", "/"), ("#ftp/", "/"), ("#news.", ".")), )) class TestCapabilities(IMAPClientTest): def test_preauth(self): self.client._imap.capabilities = ('FOO', 'BAR') self.client._imap.untagged_responses = {} self.assertEqual(self.client.capabilities(), (b'FOO', b'BAR')) def test_server_returned_capability_after_auth(self): self.client._imap.capabilities = (b'FOO',) self.client._imap.untagged_responses = {'CAPABILITY': [b'FOO MORE']} self.assertEqual(self.client._cached_capabilities, None) self.assertEqual(self.client.capabilities(), (b'FOO', b'MORE')) self.assertEqual(self.client._cached_capabilities, (b'FOO', b'MORE')) self.assertEqual(self.client._imap.untagged_responses, {}) def test_caching(self): self.client._imap.capabilities = ('FOO',) self.client._imap.untagged_responses = {} self.client._cached_capabilities = (b'FOO', b'MORE') self.assertEqual(self.client.capabilities(), (b'FOO', b'MORE')) def test_post_auth_request(self): self.client._imap.capabilities = ('FOO',) self.client._imap.untagged_responses = {} self.client._imap.state = 'SELECTED' self.client._imap.capability.return_value = ('OK', [b'FOO BAR']) self.assertEqual(self.client.capabilities(), (b'FOO', b'BAR')) self.assertEqual(self.client._cached_capabilities, (b'FOO', b'BAR')) def test_with_starttls(self): # Initial connection self.client._imap.capabilities = ('FOO',) self.client._imap.untagged_responses = {} self.client._imap.state = 'NONAUTH' self.assertEqual(self.client.capabilities(), (b'FOO',)) # Now do STARTTLS; capabilities change and should be reported. self.client._starttls_done = True self.client._imap.capability.return_value = ('OK', [b'FOO BAR']) self.assertEqual(self.client.capabilities(), (b'FOO', b'BAR')) # Login done; capabilities change again. self.client._imap.state = 'AUTH' self.client._imap.capability.return_value = ('OK', [b'FOO BAR QUX']) self.assertEqual(self.client.capabilities(), (b'FOO', b'BAR', b'QUX')) def test_has_capability(self): self.client._cached_capabilities = (b'FOO', b'MORE') self.assertTrue(self.client.has_capability(b'FOO')) self.assertTrue(self.client.has_capability(b'foo')) self.assertFalse(self.client.has_capability(b'BAR')) self.assertTrue(self.client.has_capability('FOO')) self.assertTrue(self.client.has_capability('foo')) self.assertFalse(self.client.has_capability('BAR')) class TestId(IMAPClientTest): def test_id(self): self.client._cached_capabilities = (b'ID',) self.client._imap._simple_command.return_value = ('OK', [b'Success']) self.client._imap._untagged_response.return_value = ( b'OK', [b'("name" "GImap" "vendor" "Google, Inc.")']) id_response = self.client.id_({'name': 'IMAPClient'}) self.client._imap._simple_command.assert_called_with( 'ID', '("name" "IMAPClient")') self.assertSequenceEqual( id_response, ((b'name', b'GImap', b'vendor', b'Google, Inc.'),)) def test_no_support(self): self.client._cached_capabilities = (b'IMAP4rev1',) self.assertRaises(ValueError, self.client.id_) def test_invalid_parameters(self): self.assertRaises(TypeError, self.client.id_, 'bananarama') class TestRawCommand(IMAPClientTest): def setUp(self): super(TestRawCommand, self).setUp() self.client._imap._get_response.return_value = None self.client._imap._command_complete.return_value = ('OK', ['done']) def check(self, command, args, expected): typ, data = self.client._raw_command(command, args) self.assertEqual(typ, 'OK') self.assertEqual(data, ['done']) self.assertEqual(self.client._imap.sent, expected) def test_plain(self): self.check(b'search', [b'ALL'], b'tag UID SEARCH ALL\r\n', ) def test_not_uid(self): self.client.use_uid = False self.check(b'search', [b'ALL'], b'tag SEARCH ALL\r\n', ) def test_literal_at_end(self): self.check(b'search', [b'TEXT', b'\xfe\xff'], b'tag UID SEARCH TEXT {2}\r\n' b'\xfe\xff\r\n' ) def test_embedded_literal(self): self.check(b'search', [b'TEXT', b'\xfe\xff', b'DELETED'], b'tag UID SEARCH TEXT {2}\r\n' b'\xfe\xff DELETED\r\n' ) def test_multiple_literals(self): self.check(b'search', [b'TEXT', b'\xfe\xff', b'TEXT', b'\xcc'], b'tag UID SEARCH TEXT {2}\r\n' b'\xfe\xff TEXT {1}\r\n' b'\xcc\r\n' ) def test_complex(self): self.check(b'search', [b'FLAGGED', b'TEXT', b'\xfe\xff', b'TEXT', b'\xcc', b'TEXT', b'foo'], b'tag UID SEARCH FLAGGED TEXT {2}\r\n' b'\xfe\xff TEXT {1}\r\n' b'\xcc TEXT foo\r\n' ) def test_invalid_input_type(self): self.assertRaises(ValueError, self.client._raw_command, 'foo', []) self.assertRaises(ValueError, self.client._raw_command, u'foo', ['foo']) def test_failed_continuation_wait(self): self.client._imap._get_response.return_value = b'blah' self.client._imap.tagged_commands['tag'] = ('NO', ['go away']) expected_error = "unexpected response while waiting for continuation response: \(u?'NO', \[u?'go away'\]\)" with self.assertRaisesRegex(IMAPClient.AbortError, expected_error): self.client._raw_command(b'FOO', [b'\xff']) class TestShutdown(IMAPClientTest): def test_shutdown(self): self.client.shutdown() self.client._imap.shutdown.assert_called_once_with()