Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_login_twice(self):
with self.assertRaises(aioimaplib.Error) as expected:
imap_client = yield from self.login_user('user', 'pass')
yield from imap_client.login('user', 'password')
self.assertEqual(expected.exception.args, ('command LOGIN illegal in state AUTH',))
@asyncio.coroutine
def capability(self):
response = yield from self.execute(Command('CAPABILITY', self.new_tag(), loop=self.loop))
capability_list = response.lines[0].split()
self.capabilities = set(capability_list)
version = capability_list[0].upper()
if version not in AllowedVersions:
raise Error('server not IMAP4 compliant')
else:
self.imap_version = version
if self.buffer:
self.queue.put_nowait(copy(self.buffer))
self.buffer.clear()
class AioImapException(Exception):
def __init__(self, reason):
super().__init__(reason)
class Error(AioImapException):
def __init__(self, reason):
super().__init__(reason)
class Abort(Error):
def __init__(self, reason):
super().__init__(reason)
class CommandTimeout(AioImapException):
def __init__(self, command):
self.command = command
class IncompleteRead(AioImapException):
def __init__(self, cmd, data=b''):
self.cmd = cmd
self.data = data
def change_state(coro):
@change_state
@asyncio.coroutine
def welcome(self, command):
if 'PREAUTH' in command:
self.state = AUTH
elif 'OK' in command:
self.state = NONAUTH
else:
raise Error(command)
yield from self.capability()
def _response_done(self, line):
log.debug('tagged status %s' % line)
tag, _, response = line.partition(' ')
if self.pending_sync_command is not None:
if self.pending_sync_command.tag != tag:
raise Abort('unexpected tagged response with pending sync command (%s) response: %s' %
(self.pending_sync_command, response))
command = self.pending_sync_command
self.pending_sync_command = None
else:
cmds = self._find_pending_async_cmd_by_tag(tag)
if len(cmds) == 0:
raise Abort('unexpected tagged (%s) response: %s' % (tag, response))
elif len(cmds) > 1:
raise Error('inconsistent state : two commands have the same tag (%s)' % cmds)
command = cmds.pop()
self.pending_async_commands.pop(command.untagged_resp_name)
response_result, _, response_text = response.partition(' ')
command.close(response_text, result=response_result)