mirror of
https://github.com/chatmail/core.git
synced 2026-04-17 13:36:30 +03:00
553d3936a9
This PR originally contained a fix for sqlx which turned out not not to be necessary. But on the way, I made some general improvements: - Under some circumstances, a "normal" test failure led to a timeout, without printing a decent error message. See e.g. https://app.circleci.com/pipelines/github/deltachat/deltachat-core-rust/8069/workflows/ba2a9949-b4ad-4ceb-a930-073bba05e2db/jobs/30965. (The problem was: if there is an exception in dc_account_extra_configure(), when trying to handle the exception the line account.log("===================", e, "===================") was called, which can't work as log() only expects one parameter) - When a test fails: Call `dump_account_info()` even if there is no direct_imap - In test_import_export_online_all(), add another 100KB file to the backup. This adds resilience against future size changes of the sql db file: The test tests the smoothness of the progress bar. And if there are there are not enough about-equally-sized files, the progress bar can't be smooth.
258 lines
9.0 KiB
Python
258 lines
9.0 KiB
Python
"""
|
|
Internal Python-level IMAP handling used by the testplugin
|
|
and for cleaning up inbox/mvbox for each test function run.
|
|
"""
|
|
|
|
import io
|
|
import email
|
|
import ssl
|
|
import pathlib
|
|
from imapclient import IMAPClient
|
|
from imapclient.exceptions import IMAPClientError
|
|
import imaplib
|
|
import deltachat
|
|
from deltachat import const
|
|
|
|
|
|
SEEN = b'\\Seen'
|
|
DELETED = b'\\Deleted'
|
|
FLAGS = b'FLAGS'
|
|
FETCH = b'FETCH'
|
|
ALL = "1:*"
|
|
|
|
|
|
@deltachat.global_hookimpl
|
|
def dc_account_extra_configure(account):
|
|
""" Reset the account (we reuse accounts across tests)
|
|
and make 'account.direct_imap' available for direct IMAP ops.
|
|
"""
|
|
try:
|
|
|
|
if not hasattr(account, "direct_imap"):
|
|
imap = DirectImap(account)
|
|
|
|
for folder in imap.list_folders():
|
|
if folder.lower() == "inbox" or folder.lower() == "deltachat":
|
|
assert imap.select_folder(folder)
|
|
imap.delete(ALL, expunge=True)
|
|
else:
|
|
imap.conn.delete_folder(folder)
|
|
# We just deleted the folder, so we have to make DC forget about it, too
|
|
if account.get_config("configured_sentbox_folder") == folder:
|
|
account.set_config("configured_sentbox_folder", None)
|
|
if account.get_config("configured_spam_folder") == folder:
|
|
account.set_config("configured_spam_folder", None)
|
|
|
|
setattr(account, "direct_imap", imap)
|
|
|
|
except Exception as e:
|
|
# Uncaught exceptions here would lead to a timeout without any note written to the log
|
|
# start with DC_EVENT_WARNING so that the line is printed in yellow and won't be overlooked when reading
|
|
account.log("DC_EVENT_WARNING =================== DIRECT_IMAP CAN'T RESET ACCOUNT: ===================")
|
|
account.log("DC_EVENT_WARNING =================== " + str(e) + " ===================")
|
|
|
|
|
|
@deltachat.global_hookimpl
|
|
def dc_account_after_shutdown(account):
|
|
""" shutdown the imap connection if there is one. """
|
|
imap = getattr(account, "direct_imap", None)
|
|
if imap is not None:
|
|
imap.shutdown()
|
|
del account.direct_imap
|
|
|
|
|
|
class DirectImap:
|
|
def __init__(self, account):
|
|
self.account = account
|
|
self.logid = account.get_config("displayname") or id(account)
|
|
self._idling = False
|
|
self.connect()
|
|
|
|
def connect(self):
|
|
host = self.account.get_config("configured_mail_server")
|
|
port = int(self.account.get_config("configured_mail_port"))
|
|
security = int(self.account.get_config("configured_mail_security"))
|
|
|
|
user = self.account.get_config("addr")
|
|
pw = self.account.get_config("mail_pw")
|
|
|
|
if security == const.DC_SOCKET_PLAIN:
|
|
ssl_context = None
|
|
else:
|
|
ssl_context = ssl.create_default_context()
|
|
|
|
# don't check if certificate hostname doesn't match target hostname
|
|
ssl_context.check_hostname = False
|
|
|
|
# don't check if the certificate is trusted by a certificate authority
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
if security == const.DC_SOCKET_STARTTLS:
|
|
self.conn = IMAPClient(host, port, ssl=False)
|
|
self.conn.starttls(ssl_context)
|
|
elif security == const.DC_SOCKET_PLAIN:
|
|
self.conn = IMAPClient(host, port, ssl=False)
|
|
elif security == const.DC_SOCKET_SSL:
|
|
self.conn = IMAPClient(host, port, ssl_context=ssl_context)
|
|
self.conn.login(user, pw)
|
|
|
|
self.select_folder("INBOX")
|
|
|
|
def shutdown(self):
|
|
try:
|
|
self.conn.idle_done()
|
|
except (OSError, IMAPClientError):
|
|
pass
|
|
try:
|
|
self.conn.logout()
|
|
except (OSError, IMAPClientError):
|
|
print("Could not logout direct_imap conn")
|
|
|
|
def create_folder(self, foldername):
|
|
try:
|
|
self.conn.create_folder(foldername)
|
|
except imaplib.IMAP4.error as e:
|
|
print("Can't create", foldername, "probably it already exists:", str(e))
|
|
|
|
def select_folder(self, foldername):
|
|
assert not self._idling
|
|
return self.conn.select_folder(foldername)
|
|
|
|
def select_config_folder(self, config_name):
|
|
""" Return info about selected folder if it is
|
|
configured, otherwise None. """
|
|
if "_" not in config_name:
|
|
config_name = "configured_{}_folder".format(config_name)
|
|
foldername = self.account.get_config(config_name)
|
|
if foldername:
|
|
return self.select_folder(foldername)
|
|
|
|
def list_folders(self):
|
|
""" return list of all existing folder names"""
|
|
assert not self._idling
|
|
folders = []
|
|
for meta, sep, foldername in self.conn.list_folders():
|
|
folders.append(foldername)
|
|
return folders
|
|
|
|
def delete(self, range, expunge=True):
|
|
""" delete a range of messages (imap-syntax).
|
|
If expunge is true, perform the expunge-operation
|
|
to make sure the messages are really gone and not
|
|
just flagged as deleted.
|
|
"""
|
|
self.conn.set_flags(range, [DELETED])
|
|
if expunge:
|
|
self.conn.expunge()
|
|
|
|
def get_all_messages(self):
|
|
assert not self._idling
|
|
|
|
# Flush unsolicited responses. IMAPClient has problems
|
|
# dealing with them: https://github.com/mjs/imapclient/issues/334
|
|
# When this NOOP was introduced, next FETCH returned empty
|
|
# result instead of a single message, even though IMAP server
|
|
# can only return more untagged responses than required, not
|
|
# less.
|
|
self.conn.noop()
|
|
|
|
return self.conn.fetch(ALL, [FLAGS])
|
|
|
|
def get_unread_messages(self):
|
|
assert not self._idling
|
|
res = self.conn.fetch(ALL, [FLAGS])
|
|
return [uid for uid in res
|
|
if SEEN not in res[uid][FLAGS]]
|
|
|
|
def mark_all_read(self):
|
|
messages = self.get_unread_messages()
|
|
if messages:
|
|
res = self.conn.set_flags(messages, [SEEN])
|
|
print("marked seen:", messages, res)
|
|
|
|
def get_unread_cnt(self):
|
|
return len(self.get_unread_messages())
|
|
|
|
def dump_imap_structures(self, dir, logfile):
|
|
assert not self._idling
|
|
stream = io.StringIO()
|
|
|
|
def log(*args, **kwargs):
|
|
kwargs["file"] = stream
|
|
print(*args, **kwargs)
|
|
|
|
empty_folders = []
|
|
for imapfolder in self.list_folders():
|
|
self.select_folder(imapfolder)
|
|
messages = list(self.get_all_messages())
|
|
if not messages:
|
|
empty_folders.append(imapfolder)
|
|
continue
|
|
|
|
log("---------", imapfolder, len(messages), "messages ---------")
|
|
# get message content without auto-marking it as seen
|
|
# fetching 'RFC822' would mark it as seen.
|
|
requested = [b'BODY.PEEK[]', FLAGS]
|
|
for uid, data in self.conn.fetch(messages, requested).items():
|
|
body_bytes = data[b'BODY[]']
|
|
if not body_bytes:
|
|
log("Message", uid, "has empty body")
|
|
continue
|
|
|
|
flags = data[FLAGS]
|
|
path = pathlib.Path(str(dir)).joinpath("IMAP", self.logid, imapfolder)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
fn = path.joinpath(str(uid))
|
|
fn.write_bytes(body_bytes)
|
|
log("Message", uid, fn)
|
|
email_message = email.message_from_bytes(body_bytes)
|
|
log("Message", uid, flags, "Message-Id:", email_message.get("Message-Id"))
|
|
|
|
if empty_folders:
|
|
log("--------- EMPTY FOLDERS:", empty_folders)
|
|
|
|
print(stream.getvalue(), file=logfile)
|
|
|
|
def idle_start(self):
|
|
""" switch this connection to idle mode. non-blocking. """
|
|
assert not self._idling
|
|
res = self.conn.idle()
|
|
self._idling = True
|
|
return res
|
|
|
|
def idle_check(self, terminate=False):
|
|
""" (blocking) wait for next idle message from server. """
|
|
assert self._idling
|
|
self.account.log("imap-direct: calling idle_check")
|
|
res = self.conn.idle_check(timeout=30)
|
|
if len(res) == 0:
|
|
raise TimeoutError
|
|
if terminate:
|
|
self.idle_done()
|
|
self.account.log("imap-direct: idle_check returned {!r}".format(res))
|
|
return res
|
|
|
|
def idle_wait_for_seen(self):
|
|
""" Return first message with SEEN flag
|
|
from a running idle-stream REtiurn.
|
|
"""
|
|
while 1:
|
|
for item in self.idle_check():
|
|
if item[1] == FETCH:
|
|
if item[2][0] == FLAGS:
|
|
if SEEN in item[2][1]:
|
|
return item[0]
|
|
|
|
def idle_done(self):
|
|
""" send idle-done to server if we are currently in idle mode. """
|
|
if self._idling:
|
|
res = self.conn.idle_done()
|
|
self._idling = False
|
|
return res
|
|
|
|
def append(self, folder, msg):
|
|
if msg.startswith("\n"):
|
|
msg = msg[1:]
|
|
msg = '\n'.join([s.lstrip() for s in msg.splitlines()])
|
|
self.conn.append(folder, msg)
|