apply isort and black formatters, add format checking to CI

This commit is contained in:
adbenitez
2022-05-29 21:11:49 -04:00
parent 62b50c87d4
commit 16e0f0e986
26 changed files with 899 additions and 575 deletions

View File

@@ -3,18 +3,27 @@ Internal Python-level IMAP handling used by the testplugin
and for cleaning up inbox/mvbox for each test function run.
"""
import io
import ssl
import pathlib
from contextlib import contextmanager
from imap_tools import MailBox, MailBoxTls, errors, AND, Header, MailMessageFlags, MailMessage
import imaplib
from deltachat import const, Account
import io
import pathlib
import ssl
from contextlib import contextmanager
from typing import List
from imap_tools import (
AND,
Header,
MailBox,
MailBoxTls,
MailMessage,
MailMessageFlags,
errors,
)
FLAGS = b'FLAGS'
FETCH = b'FETCH'
from deltachat import Account, const
FLAGS = b"FLAGS"
FETCH = b"FETCH"
ALL = "1:*"
@@ -69,8 +78,8 @@ class DirectImap:
return self.conn.folder.set(foldername)
def select_config_folder(self, config_name: str):
""" Return info about selected folder if it is
configured, otherwise None. """
"""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)
@@ -78,17 +87,17 @@ class DirectImap:
return self.select_folder(foldername)
def list_folders(self) -> List[str]:
""" return list of all existing folder names"""
"""return list of all existing folder names"""
assert not self._idling
return [folder.name for folder in self.conn.folder.list()]
def delete(self, uid_list: str, expunge=True):
""" delete a range of messages (imap-syntax).
"""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.client.uid('STORE', uid_list, '+FLAGS', r'(\Deleted)')
self.conn.client.uid("STORE", uid_list, "+FLAGS", r"(\Deleted)")
if expunge:
self.conn.expunge()
@@ -141,7 +150,13 @@ class DirectImap:
fn = path.joinpath(str(msg.uid))
fn.write_bytes(body)
log("Message", msg.uid, fn)
log("Message", msg.uid, msg.flags, "Message-Id:", msg.obj.get("Message-Id"))
log(
"Message",
msg.uid,
msg.flags,
"Message-Id:",
msg.obj.get("Message-Id"),
)
if empty_folders:
log("--------- EMPTY FOLDERS:", empty_folders)
@@ -150,7 +165,7 @@ class DirectImap:
@contextmanager
def idle(self):
""" return Idle ContextManager. """
"""return Idle ContextManager."""
idle_manager = IdleManager(self)
try:
yield idle_manager
@@ -163,13 +178,20 @@ class DirectImap:
"""
if msg.startswith("\n"):
msg = msg[1:]
msg = '\n'.join([s.lstrip() for s in msg.splitlines()])
self.conn.append(bytes(msg, encoding='ascii'), folder)
msg = "\n".join([s.lstrip() for s in msg.splitlines()])
self.conn.append(bytes(msg, encoding="ascii"), folder)
def get_uid_by_message_id(self, message_id) -> str:
msgs = [msg.uid for msg in self.conn.fetch(AND(header=Header('MESSAGE-ID', message_id)))]
msgs = [
msg.uid
for msg in self.conn.fetch(AND(header=Header("MESSAGE-ID", message_id)))
]
if len(msgs) == 0:
raise Exception("Did not find message " + message_id + ", maybe you forgot to select the correct folder?")
raise Exception(
"Did not find message "
+ message_id
+ ", maybe you forgot to select the correct folder?"
)
return msgs[0]
@@ -183,7 +205,7 @@ class IdleManager:
self.direct_imap.conn.idle.start()
def check(self, timeout=None) -> List[bytes]:
""" (blocking) wait for next idle message from server. """
"""(blocking) wait for next idle message from server."""
self.log("imap-direct: calling idle_check")
res = self.direct_imap.conn.idle.poll(timeout=timeout)
self.log("imap-direct: idle_check returned {!r}".format(res))
@@ -192,20 +214,19 @@ class IdleManager:
def wait_for_new_message(self, timeout=None) -> bytes:
while 1:
for item in self.check(timeout=timeout):
if b'EXISTS' in item or b'RECENT' in item:
if b"EXISTS" in item or b"RECENT" in item:
return item
def wait_for_seen(self, timeout=None) -> int:
""" Return first message with SEEN flag from a running idle-stream.
"""
"""Return first message with SEEN flag from a running idle-stream."""
while 1:
for item in self.check(timeout=timeout):
if FETCH in item:
self.log(str(item))
if FLAGS in item and rb'\Seen' in item:
return int(item.split(b' ')[1])
if FLAGS in item and rb"\Seen" in item:
return int(item.split(b" ")[1])
def done(self):
""" send idle-done to server if we are currently in idle mode. """
"""send idle-done to server if we are currently in idle mode."""
res = self.direct_imap.conn.idle.stop()
return res