GovOPlaN/govoplan-mail
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

Reuse shared IMAP message listing path

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:52:05 +02:00
parent f3faa4dff9
commit e4594c355d
2 changed files with 160 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
src/govoplan_mail/backend/sending/imap.py
View File

@@ -915,6 +915,71 @@ def _mock_raw_bytes(record: dict[str, Any]) -> bytes:
return "\n".join(lines).encode("utf-8", errors="replace")
def _mock_mailbox_page_offset(
records: list[dict[str, Any]],
*,
offset: int,
after_uid: str | None,
expected_uidvalidity: str | None,
) -> tuple[int, bool]:
if expected_uidvalidity and expected_uidvalidity != "mock-v1":
return 0, True
if after_uid:
record_ids = [str(record.get("id") or "") for record in records]
try:
return record_ids.index(str(after_uid)) + 1, False
except ValueError:
return 0, True
return min(offset, len(records)), False
def _mock_message_summary(record: dict[str, Any]) -> ImapMailboxMessageSummary:
return _message_summary_from_raw(
uid=str(record.get("id") or ""),
folder=_mock_record_folder(record),
raw=_mock_raw_bytes(record),
flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
size_bytes=int(record.get("size_bytes") or 0) or None,
)
def _list_mock_imap_messages(
*,
host: str,
port: int,
security: str,
folder: str,
limit: int,
offset: int,
after_uid: str | None,
expected_uidvalidity: str | None,
) -> ImapMailboxMessageListResult:
records = [
record
for record in list_records(limit=500)
if _mock_folder_matches(record, folder)
]
effective_offset, cursor_reset = _mock_mailbox_page_offset(
records,
offset=offset,
after_uid=after_uid,
expected_uidvalidity=expected_uidvalidity,
)
page_records = records[effective_offset : effective_offset + limit]
return ImapMailboxMessageListResult(
host=host,
port=port,
security=security,
folder=folder,
messages=[_mock_message_summary(record) for record in page_records],
total_count=len(records),
offset=effective_offset,
limit=limit,
uidvalidity="mock-v1",
cursor_reset=cursor_reset,
)
def list_imap_messages(
*,
imap_config: ImapConfig,
@@ -929,88 +994,29 @@ def list_imap_messages(
host, port = _require_imap_config(imap_config)
folder, limit, offset = _normalize_mailbox_page(folder=folder, limit=limit, offset=offset)
if is_mock_imap_host(imap_config.host):
records = [record for record in list_records(limit=500) if _mock_folder_matches(record, folder)]
cursor_reset = False
if expected_uidvalidity and expected_uidvalidity != "mock-v1":
effective_offset = 0
cursor_reset = True
elif after_uid:
record_ids = [str(record.get("id") or "") for record in records]
try:
effective_offset = record_ids.index(str(after_uid)) + 1
except ValueError:
effective_offset = 0
cursor_reset = True
else:
effective_offset = min(offset, len(records))
page_records = records[effective_offset:effective_offset + limit]
messages = [
_message_summary_from_raw(
uid=str(record.get("id") or ""),
folder=_mock_record_folder(record),
raw=_mock_raw_bytes(record),
flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
size_bytes=int(record.get("size_bytes") or 0) or None,
)
for record in page_records
]
return ImapMailboxMessageListResult(
return _list_mock_imap_messages(
host=host,
port=port,
security=imap_config.security.value,
folder=folder,
messages=messages,
total_count=len(records),
offset=effective_offset,
limit=limit,
uidvalidity="mock-v1",
cursor_reset=cursor_reset,
offset=offset,
after_uid=after_uid,
expected_uidvalidity=expected_uidvalidity,
)
client = _open_imap(imap_config)
try:
total_count, uidvalidity = _select_readonly(client, folder)
if after_uid is None and expected_uidvalidity is None:
page_sequences = _paged_descending_sequences(total_count, offset=offset, limit=limit)
messages = _fetch_message_summaries_by_sequence(client, page_sequences, folder)
return ImapMailboxMessageListResult(
return _list_imap_messages_on_client(
client,
host=host,
port=port,
security=imap_config.security.value,
folder=folder,
messages=messages,
total_count=total_count,
limit=limit,
offset=offset,
limit=limit,
uidvalidity=None,
cursor_reset=False,
)
uids = _search_message_uids(client)
cursor_reset = False
effective_after_uid = after_uid
effective_offset = offset
if expected_uidvalidity and expected_uidvalidity != uidvalidity:
effective_after_uid = None
effective_offset = 0
cursor_reset = True
page_uids, effective_offset, anchor_missing = _paged_descending_uids(
uids,
offset=effective_offset,
limit=limit,
after_uid=effective_after_uid,
)
messages = _fetch_message_summaries_by_uid(client, page_uids, folder)
return ImapMailboxMessageListResult(
host=host,
port=port,
security=imap_config.security.value,
folder=folder,
messages=messages,
total_count=len(uids) if uids else total_count,
offset=effective_offset,
limit=limit,
uidvalidity=uidvalidity,
cursor_reset=cursor_reset or anchor_missing,
after_uid=after_uid,
expected_uidvalidity=expected_uidvalidity,
)
finally:
try:

86
tests/test_imap_parser.py
View File

@@ -18,6 +18,7 @@ from govoplan_mail.backend.sending.imap import (
_select_readonly,
_sequence_set,
append_message_to_sent,
list_imap_messages,
)
@@ -79,6 +80,91 @@ class ImapFolderParserTests(unittest.TestCase):
class ImapMessagePaginationTests(unittest.TestCase):
def test_mock_message_cursor_preserves_order_and_resets_when_stale(self):
records = [
{
"id": "3",
"kind": "smtp",
"raw_eml": "Subject: Three\r\n\r\nBody",
"size_bytes": 26,
},
{
"id": "2",
"kind": "smtp",
"raw_eml": "Subject: Two\r\n\r\nBody",
"size_bytes": 24,
},
]
config = ImapConfig(host="mock.imap.local")
with patch(
"govoplan_mail.backend.sending.imap.list_records",
return_value=records,
):
page = list_imap_messages(
imap_config=config,
after_uid="3",
expected_uidvalidity="mock-v1",
limit=1,
)
reset_page = list_imap_messages(
imap_config=config,
after_uid="3",
expected_uidvalidity="stale",
limit=1,
)
self.assertEqual([message.uid for message in page.messages], ["2"])
self.assertEqual(page.offset, 1)
self.assertFalse(page.cursor_reset)
self.assertEqual([message.uid for message in reset_page.messages], ["3"])
self.assertEqual(reset_page.offset, 0)
self.assertTrue(reset_page.cursor_reset)
def test_real_message_listing_delegates_to_shared_client_path(self):
class Client:
logged_out = False
def logout(self):
self.logged_out = True
client = Client()
expected = object()
config = ImapConfig(host="imap.example.org")
with (
patch(
"govoplan_mail.backend.sending.imap._open_imap",
return_value=client,
),
patch(
"govoplan_mail.backend.sending.imap._list_imap_messages_on_client",
return_value=expected,
) as list_on_client,
):
result = list_imap_messages(
imap_config=config,
folder=" Archive ",
limit=25,
offset=5,
after_uid="42",
expected_uidvalidity="7",
)
self.assertIs(result, expected)
self.assertTrue(client.logged_out)
list_on_client.assert_called_once_with(
client,
host="imap.example.org",
port=993,
security="tls",
folder="Archive",
limit=25,
offset=5,
after_uid="42",
expected_uidvalidity="7",
)
def test_full_message_fetch_uses_partial_range_and_deployment_limit(self):
class Client:
command = ""