test(email): cover smtp_username / imap_username / case-insensitive self-address match

The original regression only exercised a from_address match with all three
identity fields set to the same value, so it couldn't distinguish whether
_self_addresses actually picks up smtp_username and imap_username or just
collapses on from_address. Add a parametrized test covering:

- smtp_username-only match (from_address empty, imap_username different) —
  simulates SMTP relays that rewrite outbound From to the login identity.
- imap_username-only match — simulates mailbox-identity setups.
- Case-insensitive match — inbound From arriving upper-cased must still hit.

No production code changes.

Made-with: Cursor
This commit is contained in:
Xubin Ren
2026-04-17 16:25:16 +08:00
committed by Xubin Ren
parent 1011ea5ac8
commit 3ae4333cef
+63
View File
@@ -132,6 +132,69 @@ def test_fetch_new_messages_skips_self_sent_email_and_marks_seen(monkeypatch) ->
assert items_again == []
@pytest.mark.parametrize(
"config_override,from_header",
[
# Only smtp_username matches — simulates an SMTP relay where
# outbound From gets rewritten to the SMTP login identity.
(
{"from_address": "", "smtp_username": "bot@example.com", "imap_username": "other@imap.com"},
"bot@example.com",
),
# Only imap_username matches — simulates mailbox-based identity
# with no explicit from_address set.
(
{"from_address": "", "smtp_username": "other@smtp.com", "imap_username": "bot@example.com"},
"bot@example.com",
),
# Case-insensitive: inbound From arrives upper-cased.
(
{"from_address": "bot@example.com", "smtp_username": "other@smtp.com", "imap_username": "other@imap.com"},
"BOT@EXAMPLE.COM",
),
],
ids=["smtp_username_only", "imap_username_only", "case_insensitive"],
)
def test_fetch_new_messages_skips_self_sent_across_identity_sources(
monkeypatch, config_override, from_header
) -> None:
"""Self-address detection must fire when any of from_address / smtp_username /
imap_username matches, and must be case-insensitive."""
raw = _make_raw_email(from_addr=from_header, subject="Loop test")
class FakeIMAP:
def __init__(self) -> None:
self.store_calls: list[tuple[bytes, str, str]] = []
def login(self, _user: str, _pw: str):
return "OK", [b"logged in"]
def select(self, _mailbox: str):
return "OK", [b"1"]
def search(self, *_args):
return "OK", [b"1"]
def fetch(self, _imap_id: bytes, _parts: str):
return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
def store(self, imap_id: bytes, op: str, flags: str):
self.store_calls.append((imap_id, op, flags))
return "OK", [b""]
def logout(self):
return "BYE", [b""]
fake = FakeIMAP()
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
channel = EmailChannel(_make_config(**config_override), MessageBus())
items = channel._fetch_new_messages()
assert items == []
assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
def test_fetch_new_messages_retries_once_when_imap_connection_goes_stale(monkeypatch) -> None:
raw = _make_raw_email(subject="Invoice", body="Please pay")
fail_once = {"pending": True}