html body
"): # Build with the legacy MIME* constructors, then round-trip # through email.message_from_bytes with the default policy so the # parts are EmailMessage instances with .get_content() — matching # what cmd_read sees when imaplib hands it RFC822 bytes. msg = MIMEMultipart("alternative") if plain is not None: msg.attach(MIMEText(plain, "plain")) if html is not None: msg.attach(MIMEText(html, "html")) return email.message_from_bytes(msg.as_bytes(), policy=default_policy) def test_normal_multipart_prefers_text_plain(self, cmail_action): msg = self._multipart_alt(plain="plain wins", html="html loses
") assert cmail_action.extract_body(msg) == "plain wins" def test_boundary_html_only_multipart_falls_back_to_html(self, cmail_action): msg = self._multipart_alt(plain=None, html="only html
") result = cmail_action.extract_body(msg) assert result is not None assert "only html" in result def test_boundary_singlepart_returns_content_directly(self, cmail_action): msg = EmailMessage() msg.set_content("single-part body") # set_content adds Content-Type: text/plain by default; result has # a trailing newline from the policy formatter. assert cmail_action.extract_body(msg).strip() == "single-part body" def test_error_multipart_with_no_text_parts_returns_none(self, cmail_action): msg = MIMEMultipart("alternative") msg.attach(MIMEApplication(b"binary blob")) # Round-trip for parity with the parser-based path real callers use. parsed = email.message_from_bytes(msg.as_bytes(), policy=default_policy) assert cmail_action.extract_body(parsed) is None # --------------------------------------------------------------------------- # _decode_header — pure # --------------------------------------------------------------------------- class TestDecodeHeader: def test_normal_string(self, cmail_action): assert cmail_action._decode_header("hello") == "hello" def test_boundary_empty_string(self, cmail_action): assert cmail_action._decode_header("") == "" def test_boundary_none_returns_empty(self, cmail_action): assert cmail_action._decode_header(None) == "" def test_boundary_non_string_coerced_via_str(self, cmail_action): assert cmail_action._decode_header(42) == "42" # --------------------------------------------------------------------------- # Helpers for I/O command tests # --------------------------------------------------------------------------- def _build_fetch_response(uid, from_addr="alice@example.com", subject="Hello", size=1500): """Mimic imaplib's FETCH response shape: BODY literal as a tuple, trailing FLAGS/SIZE/close-paren as a separate bytes chunk. """ headers = ( f"From: {from_addr}\r\n" f"To: c@cjennings.net\r\n" f"Subject: {subject}\r\n" f"Date: Thu, 07 May 2026 12:00:00 -0500\r\n" ).encode() return ("OK", [ (f"{uid} (BODY[HEADER.FIELDS (FROM TO SUBJECT DATE)] " f"{{{len(headers)}}}".encode(), headers), f" FLAGS () RFC822.SIZE {size})".encode(), ]) # --------------------------------------------------------------------------- # cmd_list_unread — mocked imaplib # --------------------------------------------------------------------------- class TestCmdListUnread: def test_normal_three_unread(self, cmail_action, capsys): fetch_responses = { b"100": _build_fetch_response("100", "alice@example.com", "Hello", 1500), b"101": _build_fetch_response("101", "bob@example.com", "Howdy", 2000), b"102": _build_fetch_response("102", "carol@example.com", "Hi", 500), } def uid_side_effect(cmd, *args): if cmd == "SEARCH": return ("OK", [b"100 101 102"]) if cmd == "FETCH": return fetch_responses[args[0]] return ("OK", [b""]) mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = uid_side_effect with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_list_unread(SimpleNamespace(limit=50)) parsed = json.loads(capsys.readouterr().out) assert len(parsed) == 3 assert parsed[0]["uid"] == "100" assert parsed[0]["from"] == "alice@example.com" assert parsed[0]["subject"] == "Hello" assert parsed[0]["size"] == 1500 assert parsed[2]["uid"] == "102" def test_boundary_zero_unread(self, cmail_action, capsys): mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = lambda cmd, *a: ("OK", [b""]) with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_list_unread(SimpleNamespace(limit=50)) assert json.loads(capsys.readouterr().out) == [] def test_boundary_single_unread(self, cmail_action, capsys): def uid_se(cmd, *args): if cmd == "SEARCH": return ("OK", [b"42"]) if cmd == "FETCH": return _build_fetch_response("42", "x@y", "Solo", 100) return ("OK", [b""]) mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = uid_se with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_list_unread(SimpleNamespace(limit=50)) parsed = json.loads(capsys.readouterr().out) assert len(parsed) == 1 assert parsed[0]["uid"] == "42" def test_boundary_limit_truncates_to_most_recent(self, cmail_action, capsys): # 10 unread, limit=3 — keeps the last 3 (most recent). all_uids = [str(i).encode() for i in range(100, 110)] def uid_se(cmd, *args): if cmd == "SEARCH": return ("OK", [b" ".join(all_uids)]) if cmd == "FETCH": return _build_fetch_response(args[0].decode(), "x@y", "S", 100) return ("OK", [b""]) mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = uid_se with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_list_unread(SimpleNamespace(limit=3)) parsed = json.loads(capsys.readouterr().out) assert [p["uid"] for p in parsed] == ["107", "108", "109"] def test_error_search_returns_no(self, cmail_action): mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.return_value = ("NO", [b"server error"]) with patch.object(cmail_action, "connect", return_value=mock_imap): with pytest.raises(SystemExit): cmail_action.cmd_list_unread(SimpleNamespace(limit=50)) # --------------------------------------------------------------------------- # cmd_read — mocked imaplib # --------------------------------------------------------------------------- class TestCmdRead: @staticmethod def _rfc822(body="hello world", subject="Test"): msg = EmailMessage() msg["From"] = "alice@example.com" msg["To"] = "c@cjennings.net" msg["Subject"] = subject msg["Date"] = "Thu, 07 May 2026 12:00:00 -0500" msg.set_content(body) return bytes(msg) def test_normal_prints_headers_and_body(self, cmail_action, capsys): raw = self._rfc822(body="body content here", subject="subj") mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.return_value = ("OK", [(b"1 (RFC822 {N}", raw)]) with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_read(SimpleNamespace(uid=42)) out = capsys.readouterr().out assert "From: alice@example.com" in out assert "Subject: subj" in out assert "body content here" in out def test_error_uid_not_found(self, cmail_action): mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) # imaplib's shape when the UID has no match: ('OK', [None]) mock_imap.uid.return_value = ("OK", [None]) with patch.object(cmail_action, "connect", return_value=mock_imap): with pytest.raises(SystemExit): cmail_action.cmd_read(SimpleNamespace(uid=999999)) # --------------------------------------------------------------------------- # _store wrappers — STORE command shape verification # --------------------------------------------------------------------------- class TestStoreCommands: @staticmethod def _capture_calls(cmail_action, cmd_func, uids, store_typ="OK"): calls = [] mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) def uid_se(cmd, uid, op, flags): calls.append((cmd, op, flags)) return (store_typ, [b""]) mock_imap.uid.side_effect = uid_se with patch.object(cmail_action, "connect", return_value=mock_imap): cmd_func(SimpleNamespace(uids=uids)) return calls def test_normal_mark_read_uses_plus_seen(self, cmail_action): calls = self._capture_calls(cmail_action, cmail_action.cmd_mark_read, [42]) assert calls == [("STORE", "+FLAGS", r"(\Seen)")] def test_normal_mark_unread_uses_minus_seen(self, cmail_action): calls = self._capture_calls(cmail_action, cmail_action.cmd_mark_unread, [42]) assert calls == [("STORE", "-FLAGS", r"(\Seen)")] def test_normal_star_uses_plus_flagged_and_seen(self, cmail_action): calls = self._capture_calls(cmail_action, cmail_action.cmd_star, [42]) assert calls == [("STORE", "+FLAGS", r"(\Flagged \Seen)")] def test_normal_unstar_uses_minus_flagged(self, cmail_action): calls = self._capture_calls(cmail_action, cmail_action.cmd_unstar, [42]) assert calls == [("STORE", "-FLAGS", r"(\Flagged)")] def test_boundary_multi_uid_calls_store_per_uid(self, cmail_action): calls = self._capture_calls( cmail_action, cmail_action.cmd_mark_read, [1, 2, 3] ) assert len(calls) == 3 assert all(c == ("STORE", "+FLAGS", r"(\Seen)") for c in calls) def test_error_store_failure_raises_systemexit(self, cmail_action): with pytest.raises(SystemExit): self._capture_calls( cmail_action, cmail_action.cmd_mark_read, [42], store_typ="NO" ) # --------------------------------------------------------------------------- # cmd_trash — MOVE happy path + COPY+DELETE+EXPUNGE fallback # --------------------------------------------------------------------------- class TestCmdTrash: def test_normal_move_succeeds_and_expunges(self, cmail_action): mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.return_value = ("OK", [b""]) mock_imap.expunge.return_value = ("OK", [b""]) with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_trash(SimpleNamespace(uids=[100, 101])) move_calls = [c for c in mock_imap.uid.call_args_list if c[0][0] == "MOVE"] assert len(move_calls) == 2 assert mock_imap.expunge.called def test_boundary_move_fails_falls_back_to_copy_then_delete(self, cmail_action): # MOVE returns NO -> fallback path: COPY, then STORE +FLAGS \Deleted, # then EXPUNGE. Verify the sequence executes as documented. seen_cmds = [] def uid_se(cmd, *args): seen_cmds.append(cmd) if cmd == "MOVE": return ("NO", [b"not supported"]) return ("OK", [b""]) mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = uid_se mock_imap.expunge.return_value = ("OK", [b""]) with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_trash(SimpleNamespace(uids=[100])) assert seen_cmds == ["MOVE", "COPY", "STORE"] assert mock_imap.expunge.called def test_error_copy_also_fails(self, cmail_action): mock_imap = MagicMock() mock_imap.select.return_value = ("OK", [b""]) mock_imap.uid.side_effect = lambda cmd, *a: ("NO", [b"both fail"]) with patch.object(cmail_action, "connect", return_value=mock_imap): with pytest.raises(SystemExit): cmail_action.cmd_trash(SimpleNamespace(uids=[100])) # --------------------------------------------------------------------------- # cmd_folders # --------------------------------------------------------------------------- class TestCmdFolders: def test_normal_lists_folders(self, cmail_action, capsys): mock_imap = MagicMock() mock_imap.list.return_value = ("OK", [ b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren \\Trash) "/" "Trash"', ]) with patch.object(cmail_action, "connect", return_value=mock_imap): cmail_action.cmd_folders(SimpleNamespace()) out = capsys.readouterr().out assert "INBOX" in out assert "Trash" in out def test_error_list_returns_no(self, cmail_action): mock_imap = MagicMock() mock_imap.list.return_value = ("NO", [b"server error"]) with patch.object(cmail_action, "connect", return_value=mock_imap): with pytest.raises(SystemExit): cmail_action.cmd_folders(SimpleNamespace()) # --------------------------------------------------------------------------- # build_message — pure # --------------------------------------------------------------------------- class TestBuildMessage: def test_normal_no_attachments_is_singlepart(self, cmail_action): msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="recipient@example.com", subject="Hello", body="hello world", ) assert msg["From"] == "c@cjennings.net" assert msg["To"] == "recipient@example.com" assert msg["Subject"] == "Hello" assert not msg.is_multipart() assert msg.get_content().strip() == "hello world" assert msg.get_content_type() == "text/plain" def test_normal_one_attachment_makes_multipart(self, cmail_action): attachment = ("report.txt", "text", "plain", b"line1\nline2\n") msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="recipient@example.com", subject="With file", body="see attached", attachments=[attachment], ) assert msg.is_multipart() # Find the attachment part by Content-Disposition. attached_parts = [ p for p in msg.iter_attachments() if p.get_filename() == "report.txt" ] assert len(attached_parts) == 1 att = attached_parts[0] assert att.get_content_type() == "text/plain" assert att.get_content().rstrip("\n") == "line1\nline2" def test_boundary_two_attachments(self, cmail_action): atts = [ ("a.txt", "text", "plain", b"alpha"), ("b.bin", "application", "octet-stream", b"\x00\x01\x02"), ] msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="recipient@example.com", subject="Two files", body="see attached", attachments=atts, ) names = sorted(p.get_filename() for p in msg.iter_attachments()) assert names == ["a.txt", "b.bin"] def test_boundary_empty_body(self, cmail_action): msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="recipient@example.com", subject="Empty", body="", ) # Body part exists, content is empty (modulo trailing newline). assert msg.get_content().strip() == "" def test_boundary_unicode_preserved_through_serialization(self, cmail_action): msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="recipient@example.com", subject="日本語 ñ ü", body="café — naïve résumé", ) # Round-trip: serialize, parse, check both Subject and body survived. raw = msg.as_bytes() parsed = email.message_from_bytes(raw, policy=default_policy) assert parsed["Subject"] == "日本語 ñ ü" assert "café" in parsed.get_content() def test_cc_and_bcc_headers_set_from_lists(self, cmail_action): msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="to@example.com", subject="Re: thread", body="body", cc=["cc1@example.com", "cc2@example.com"], bcc=["bcc@example.com"], ) assert msg["Cc"] == "cc1@example.com, cc2@example.com" assert msg["Bcc"] == "bcc@example.com" def test_threading_headers_set(self, cmail_action): msg = cmail_action.build_message( from_addr="c@cjennings.net", to_addr="to@example.com", subject="Re: thread", body="body", in_reply_to="