|
| 1 | +import email |
1 | 2 | import imaplib |
2 | 3 | import ipaddress |
3 | 4 | import smtplib |
4 | 5 | import ssl |
5 | 6 | from email.mime.text import MIMEText |
6 | 7 |
|
7 | | -import imap_tools |
8 | 8 | import pytest |
9 | 9 | import requests |
10 | 10 |
|
@@ -75,8 +75,11 @@ def test_delivery_port_blocked( |
75 | 75 | relayadmin2.block_port(443) |
76 | 76 | relayadmin2.block_port(80) |
77 | 77 | chat.send_text("should not arrive") |
78 | | - # Postfix logs "status=deferred", madmail logs "delivery attempt failed". |
79 | | - lines = relayadmin.wait_for_journal_match("deferred|delivery attempt failed") |
| 78 | + # Postfix logs "status=deferred"; madmail logs one of three phrasings |
| 79 | + # depending on retry state (see chatmail-delivery/src/queue/worker.rs). |
| 80 | + lines = relayadmin.wait_for_journal_match( |
| 81 | + "deferred|outbound delivery (failed, requeued|permanent failure|exceeded max_tries)" |
| 82 | + ) |
80 | 83 | lp.indent(lines.splitlines()[-1]) |
81 | 84 |
|
82 | 85 | def test_one_on_one_http_only(self, cmfactory, cmfactory2, relayadmin2, lp): |
@@ -115,11 +118,19 @@ def test_hide_senders_ip_address(cmfactory, ssl_context): |
115 | 118 | addr = user2.get_config("addr") |
116 | 119 | host = addr.split("@")[1].strip("[]") |
117 | 120 | pw = user2.get_config("mail_pw") |
118 | | - mailbox = imap_tools.MailBox(host, ssl_context=ssl_context) |
119 | | - mailbox.login(addr, pw) |
120 | | - msgs = list(mailbox.fetch(mark_seen=False)) |
121 | | - assert msgs, "expected at least one message" |
122 | | - assert public_ip not in msgs[0].obj.as_string() |
| 121 | + |
| 122 | + # madmail's IMAP server doesn't implement SEARCH/UID SEARCH (only FETCH, |
| 123 | + # UID FETCH/STORE/MOVE/COPY, see chatmail-imap/src/session.rs), so |
| 124 | + # fetch by sequence range instead. |
| 125 | + imap = imaplib.IMAP4_SSL(host, ssl_context=ssl_context) |
| 126 | + imap.login(addr, pw) |
| 127 | + imap.select("INBOX") |
| 128 | + typ, data = imap.fetch("1:*", "(BODY.PEEK[])") |
| 129 | + assert typ == "OK" |
| 130 | + raw_messages = [item[1] for item in data if isinstance(item, tuple)] |
| 131 | + assert raw_messages, "expected at least one message" |
| 132 | + msg = email.message_from_bytes(raw_messages[-1]) |
| 133 | + assert public_ip not in msg.as_string() |
123 | 134 |
|
124 | 135 |
|
125 | 136 | def test_unencrypted_rejection(cmsetup, lp): |
|
0 commit comments