return self
def __exit__(self, *args):
+ if self.state == "LOGOUT":
+ return
+
try:
self.logout()
except OSError:
Returns server 'BYE' response.
"""
self.state = 'LOGOUT'
- try: typ, dat = self._simple_command('LOGOUT')
- except: typ, dat = 'NO', ['%s: %s' % sys.exc_info()[:2]]
+ typ, dat = self._simple_command('LOGOUT')
self.shutdown()
- if 'BYE' in self.untagged_responses:
- return 'BYE', self.untagged_responses['BYE']
return typ, dat
def _command_complete(self, name, tag):
+ logout = (name == 'LOGOUT')
# BYE is expected after LOGOUT
- if name != 'LOGOUT':
+ if not logout:
self._check_bye()
try:
- typ, data = self._get_tagged_response(tag)
+ typ, data = self._get_tagged_response(tag, expect_bye=logout)
except self.abort as val:
raise self.abort('command: %s => %s' % (name, val))
except self.error as val:
raise self.error('command: %s => %s' % (name, val))
- if name != 'LOGOUT':
+ if not logout:
self._check_bye()
if typ == 'BAD':
raise self.error('%s command error: %s %s' % (name, typ, data))
return resp
- def _get_tagged_response(self, tag):
+ def _get_tagged_response(self, tag, expect_bye=False):
while 1:
result = self.tagged_commands[tag]
del self.tagged_commands[tag]
return result
+ if expect_bye:
+ typ = 'BYE'
+ bye = self.untagged_responses.pop(typ, None)
+ if bye is not None:
+ # Server replies to the "LOGOUT" command with "BYE"
+ return (typ, bye)
+
# If we've seen a BYE at this point, the socket will be
# closed, so report the BYE now.
-
self._check_bye()
# Some have reported "unexpected response" exceptions.
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(typ, 'BYE', (typ, data))
+ self.assertEqual(data[0], b'IMAP4ref1 Server logging out', (typ, data))
self.assertEqual(client.state, 'LOGOUT')
def test_lsub(self):
with transient_internet(self.host):
rs = self.server.logout()
self.server = None
- self.assertEqual(rs[0], 'BYE')
+ self.assertEqual(rs[0], 'BYE', rs)
@unittest.skipUnless(ssl, "SSL not available")
with transient_internet(self.host):
_server = self.imap_class(self.host, self.port)
rs = _server.logout()
- self.assertEqual(rs[0], 'BYE')
+ self.assertEqual(rs[0], 'BYE', rs)
def test_ssl_context_certfile_exclusive(self):
with transient_internet(self.host):