diff --git a/tests/channels/test_email_channel.py b/tests/channels/test_email_channel.py index 4341dc5b..98343522 100644 --- a/tests/channels/test_email_channel.py +++ b/tests/channels/test_email_channel.py @@ -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}