From 63a76f6bd6cdc3a6cfc584997d4519f82aa7431a Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sat, 25 Apr 2026 23:37:26 -0700 Subject: [PATCH] feat(ai_mail): test: cover all 44 untested functions per TDPLAN-0003 Co-Authored-By: @ai_mail --- src/aipass/ai_mail/tests/test_close_ops.py | 197 ++++ src/aipass/ai_mail/tests/test_create.py | 214 ++++ src/aipass/ai_mail/tests/test_email_module.py | 1034 +++++++++++++++++ .../ai_mail/tests/test_error_dispatch.py | 205 ++++ src/aipass/ai_mail/tests/test_footer.py | 88 ++ src/aipass/ai_mail/tests/test_format.py | 284 +++++ src/aipass/ai_mail/tests/test_header.py | 114 ++ .../ai_mail/tests/test_inbox_cleanup.py | 315 +++++ .../ai_mail/tests/test_misc_handlers.py | 536 +++++++++ src/aipass/ai_mail/tests/test_purge.py | 202 ++++ src/aipass/ai_mail/tests/test_reply.py | 270 +++++ src/aipass/ai_mail/tests/test_send_helpers.py | 359 ++++++ 12 files changed, 3818 insertions(+) create mode 100644 src/aipass/ai_mail/tests/test_close_ops.py create mode 100644 src/aipass/ai_mail/tests/test_create.py create mode 100644 src/aipass/ai_mail/tests/test_email_module.py create mode 100644 src/aipass/ai_mail/tests/test_error_dispatch.py create mode 100644 src/aipass/ai_mail/tests/test_footer.py create mode 100644 src/aipass/ai_mail/tests/test_format.py create mode 100644 src/aipass/ai_mail/tests/test_header.py create mode 100644 src/aipass/ai_mail/tests/test_inbox_cleanup.py create mode 100644 src/aipass/ai_mail/tests/test_misc_handlers.py create mode 100644 src/aipass/ai_mail/tests/test_purge.py create mode 100644 src/aipass/ai_mail/tests/test_reply.py create mode 100644 src/aipass/ai_mail/tests/test_send_helpers.py diff --git a/src/aipass/ai_mail/tests/test_close_ops.py b/src/aipass/ai_mail/tests/test_close_ops.py new file mode 100644 index 00000000..a4cdb485 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_close_ops.py @@ -0,0 +1,197 @@ +# =================== AIPass ==================== +# Name: test_close_ops.py +# Description: Tests for email close operations handler +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for email close operations handler -- batch close and post-ops.""" + +import pytest +from pathlib import Path +from unittest.mock import MagicMock + +import aipass.ai_mail.apps.handlers.email.close_ops as mod + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(monkeypatch): + """Prevent log_operation from writing real JSON files during tests.""" + mock_jh = MagicMock() + mock_jh.log_operation.return_value = True + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +# ---- batch_close tests ---------------------------------------- + + +def test_batch_close_single_message_success(tmp_path: Path): + """Single message close calls mark_closed_fn without skip_post_ops.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + mock_fn = MagicMock(return_value=(True, "Closed msg-1")) + + results, closed, failed = mod.batch_close(branch_path, ["msg-1"], mock_fn) + + assert len(results) == 1 + assert results[0] == ("msg-1", True, "Closed msg-1") + assert closed == 1 + assert failed == 0 + # Single message: skip_post_ops should be False + mock_fn.assert_called_once_with(branch_path, "msg-1", skip_post_ops=False) + + +def test_batch_close_multiple_messages_skip_post_ops(tmp_path: Path): + """Multiple messages pass skip_post_ops=True to mark_closed_fn.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + mock_fn = MagicMock(return_value=(True, "Closed")) + + results, closed, failed = mod.batch_close(branch_path, ["msg-1", "msg-2", "msg-3"], mock_fn) + + assert len(results) == 3 + assert closed == 3 + assert failed == 0 + # All calls should have skip_post_ops=True for batch mode + for call in mock_fn.call_args_list: + assert call.kwargs["skip_post_ops"] is True + + +def test_batch_close_mixed_results(tmp_path: Path): + """Mixed success/failure results are counted correctly.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + def _side_effect(_bp: Path, msg_id: str, skip_post_ops: bool = False): + if msg_id == "msg-2": + return False, "Not found" + return True, f"Closed {msg_id}" + + mock_fn = MagicMock(side_effect=_side_effect) + + results, closed, failed = mod.batch_close(branch_path, ["msg-1", "msg-2", "msg-3"], mock_fn) + + assert len(results) == 3 + assert closed == 2 + assert failed == 1 + assert results[1] == ("msg-2", False, "Not found") + + +def test_batch_close_empty_list(tmp_path: Path): + """Empty message list returns empty results.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + mock_fn = MagicMock() + + results, closed, failed = mod.batch_close(branch_path, [], mock_fn) + + assert results == [] + assert closed == 0 + assert failed == 0 + mock_fn.assert_not_called() + + +def test_batch_close_all_failures(tmp_path: Path): + """All failures increment failed_count, closed_count stays 0.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + mock_fn = MagicMock(return_value=(False, "Error")) + + results, closed, failed = mod.batch_close(branch_path, ["msg-1", "msg-2"], mock_fn) + + assert closed == 0 + assert failed == 2 + + +# ---- batch_close_post_ops tests -------------------------------- + + +def test_batch_close_post_ops_all_fns_called(tmp_path: Path): + """All provided functions are called with correct arguments.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + push_fn = MagicMock() + central_fn = MagicMock() + purge_fn = MagicMock() + + mod.batch_close_post_ops(branch_path, push_fn, central_fn, purge_fn) + + push_fn.assert_called_once_with(branch_path) + central_fn.assert_called_once_with() + purge_fn.assert_called_once_with(branch_path / ".ai_mail.local") + + +def test_batch_close_post_ops_none_fns(tmp_path: Path): + """None functions are skipped without error.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + # Should not raise + mod.batch_close_post_ops(branch_path, None, None, None) + + +def test_batch_close_post_ops_push_exception_suppressed(tmp_path: Path): + """Exception in push_dashboard_fn is caught; other fns still called.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + push_fn = MagicMock(side_effect=RuntimeError("push failed")) + central_fn = MagicMock() + purge_fn = MagicMock() + + mod.batch_close_post_ops(branch_path, push_fn, central_fn, purge_fn) + + central_fn.assert_called_once() + purge_fn.assert_called_once() + + +def test_batch_close_post_ops_central_exception_suppressed(tmp_path: Path): + """Exception in update_central_fn is caught; purge still called.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + push_fn = MagicMock() + central_fn = MagicMock(side_effect=RuntimeError("central failed")) + purge_fn = MagicMock() + + mod.batch_close_post_ops(branch_path, push_fn, central_fn, purge_fn) + + push_fn.assert_called_once() + purge_fn.assert_called_once() + + +def test_batch_close_post_ops_purge_exception_suppressed(tmp_path: Path): + """Exception in purge_deleted_fn is caught silently.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + push_fn = MagicMock() + central_fn = MagicMock() + purge_fn = MagicMock(side_effect=RuntimeError("purge failed")) + + mod.batch_close_post_ops(branch_path, push_fn, central_fn, purge_fn) + + push_fn.assert_called_once() + central_fn.assert_called_once() + + +def test_batch_close_post_ops_partial_fns(tmp_path: Path): + """Only provided functions are called; others default to None.""" + branch_path = tmp_path / "branch" + branch_path.mkdir() + + central_fn = MagicMock() + + mod.batch_close_post_ops(branch_path, None, central_fn, None) + + central_fn.assert_called_once_with() diff --git a/src/aipass/ai_mail/tests/test_create.py b/src/aipass/ai_mail/tests/test_create.py new file mode 100644 index 00000000..b14f0689 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_create.py @@ -0,0 +1,214 @@ +# =================== AIPass ==================== +# Name: test_create.py +# Description: Tests for email file creation handler +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for email file creation handler -- create_email_file, load_email_file.""" + +import json + +import pytest +from pathlib import Path +from unittest.mock import MagicMock + +import aipass.ai_mail.apps.handlers.email.create as mod + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(monkeypatch): + """Prevent log_operation from writing real JSON files during tests.""" + mock_jh = MagicMock() + mock_jh.log_operation.return_value = True + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +@pytest.fixture(autouse=True) +def _mock_append_footer(monkeypatch): + """Replace _get_append_footer so it returns message unchanged.""" + monkeypatch.setattr(mod, "_get_append_footer", lambda: lambda msg: msg) + + +@pytest.fixture(autouse=True) +def _mock_trigger_sent_purge(monkeypatch): + """Replace _trigger_sent_purge with a no-op.""" + monkeypatch.setattr(mod, "_trigger_sent_purge", lambda _path: None) + + +def _make_user_info(tmp_path: Path) -> dict: + """Build a minimal user_info dict pointing at tmp_path as mailbox.""" + return { + "email_address": "test@branch", + "display_name": "Test Branch", + "timestamp_format": "%Y-%m-%d %H:%M:%S", + "mailbox_path": str(tmp_path / ".ai_mail.local"), + } + + +# ---- create_email_file tests ---------------------------------- + + +def test_create_email_file_returns_path(tmp_path: Path): + """create_email_file returns a Path inside the sent/ folder.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@admin", + subject="Hello", + message="Test body", + user_info=user_info, + ) + + assert isinstance(result, Path) + assert result.exists() + assert result.parent.name == "sent" + + +def test_create_email_file_json_content(tmp_path: Path): + """Created file contains correct JSON fields.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@admin", + subject="Test Subject", + message="Body text", + user_info=user_info, + ) + + with open(result, "r", encoding="utf-8") as f: + data = json.load(f) + + assert data["from"] == "test@branch" + assert data["from_name"] == "Test Branch" + assert data["to"] == "@admin" + assert data["subject"] == "Test Subject" + assert data["message"] == "Body text" + assert data["status"] == "sent" + assert "timestamp" in data + + +def test_create_email_file_with_reply_to(tmp_path: Path): + """reply_to field is included when provided.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@worker", + subject="Task", + message="Do this", + user_info=user_info, + reply_to="@manager", + ) + + with open(result, "r", encoding="utf-8") as f: + data = json.load(f) + + assert data["reply_to"] == "@manager" + + +def test_create_email_file_with_dispatched_to(tmp_path: Path): + """dispatched_to field is included when provided.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@worker", + subject="Reply", + message="Got it", + user_info=user_info, + dispatched_to="@original", + ) + + with open(result, "r", encoding="utf-8") as f: + data = json.load(f) + + assert data["dispatched_to"] == "@original" + + +def test_create_email_file_no_optional_fields(tmp_path: Path): + """Without reply_to/dispatched_to, those keys are absent from JSON.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@admin", + subject="Plain", + message="Just a message", + user_info=user_info, + ) + + with open(result, "r", encoding="utf-8") as f: + data = json.load(f) + + assert "reply_to" not in data + assert "dispatched_to" not in data + + +def test_create_email_file_safe_filename(tmp_path: Path): + """Special characters in subject are replaced in the filename.""" + user_info = _make_user_info(tmp_path) + + result = mod.create_email_file( + to_branch="@admin", + subject="Hello/World: Test!", + message="Body", + user_info=user_info, + ) + + # Filename should not contain / or : or ! + assert "/" not in result.name.replace("/", "") + assert ":" not in result.name + assert "!" not in result.name + assert result.name.endswith(".json") + + +def test_create_email_file_creates_sent_dir(tmp_path: Path): + """sent/ directory is created if it does not exist.""" + user_info = _make_user_info(tmp_path) + sent_dir = Path(user_info["mailbox_path"]) / "sent" + assert not sent_dir.exists() + + mod.create_email_file( + to_branch="@admin", + subject="First", + message="Body", + user_info=user_info, + ) + + assert sent_dir.is_dir() + + +# ---- load_email_file tests ------------------------------------ + + +def test_load_email_file_valid(tmp_path: Path): + """load_email_file returns dict for valid JSON file.""" + email_file = tmp_path / "test_email.json" + data = {"from": "test@branch", "subject": "Hello", "message": "Body"} + email_file.write_text(json.dumps(data), encoding="utf-8") + + result = mod.load_email_file(email_file) + + assert result is not None + assert result["from"] == "test@branch" + assert result["subject"] == "Hello" + + +def test_load_email_file_missing(tmp_path: Path): + """load_email_file returns None for nonexistent file.""" + result = mod.load_email_file(tmp_path / "does_not_exist.json") + + assert result is None + + +def test_load_email_file_invalid_json(tmp_path: Path): + """load_email_file returns None for invalid JSON content.""" + email_file = tmp_path / "bad.json" + email_file.write_text("not valid json {{{", encoding="utf-8") + + result = mod.load_email_file(email_file) + + assert result is None diff --git a/src/aipass/ai_mail/tests/test_email_module.py b/src/aipass/ai_mail/tests/test_email_module.py new file mode 100644 index 00000000..bf7c8a24 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_email_module.py @@ -0,0 +1,1034 @@ +# =================== AIPass ==================== +# Name: test_email_module.py +# Description: Tests for email.py and email_send.py orchestrator functions +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for email.py and email_send.py orchestrator functions. + +Covers: handle_inbox, handle_view, handle_close, handle_reply, +handle_sent, handle_contacts, handle_register (email.py), +and handle_send (email_send.py). + +All handler dependencies are mocked -- these tests verify orchestration +logic, not business logic. +""" + +import json +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Autouse fixture: suppress json_handler.log_operation across both modules +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _silence_json_handler(): + """Prevent log_operation from writing real JSON files during tests.""" + with ( + patch("aipass.ai_mail.apps.modules.email.json_handler") as mock_email_jh, + patch("aipass.ai_mail.apps.modules.email_send.json_handler") as mock_send_jh, + ): + mock_email_jh.log_operation.return_value = True + mock_send_jh.log_operation.return_value = True + yield + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + + +def _write_inbox(tmp_path: Path, messages: list | None = None) -> Path: + """Create a minimal .ai_mail.local/inbox.json under tmp_path.""" + mailbox = tmp_path / ".ai_mail.local" + mailbox.mkdir(parents=True, exist_ok=True) + inbox_file = mailbox / "inbox.json" + data = {"messages": messages or []} + inbox_file.write_text(json.dumps(data), encoding="utf-8") + return inbox_file + + +# =========================================================================== +# handle_inbox +# =========================================================================== + + +class TestHandleInbox: + """Tests for email.handle_inbox orchestrator.""" + + def test_inbox_empty_messages(self, tmp_path, monkeypatch): + """Empty inbox prints 'is empty' message.""" + _write_inbox(tmp_path, messages=[]) + + printed: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.resolve_inbox_target", + lambda first_arg, repo_root, get_branch_fn, get_user_fn: ( + True, + { + "inbox_file": tmp_path / ".ai_mail.local" / "inbox.json", + "display_name": "TEST", + "target_branch": None, + "error": None, + }, + ), + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.load_inbox", + lambda f: {"messages": []}, + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_inbox + + result = handle_inbox([]) + assert result is True + assert any("empty" in p.lower() for p in printed) + + def test_inbox_with_messages(self, tmp_path, monkeypatch): + """Inbox with messages formats and displays them.""" + messages = [ + {"id": "m1", "status": "new", "subject": "Hello"}, + {"id": "m2", "status": "opened", "subject": "World"}, + ] + _write_inbox(tmp_path, messages=messages) + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.resolve_inbox_target", + lambda first_arg, repo_root, get_branch_fn, get_user_fn: ( + True, + { + "inbox_file": tmp_path / ".ai_mail.local" / "inbox.json", + "display_name": "TEST", + "target_branch": None, + "error": None, + }, + ), + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.load_inbox", + lambda f: {"messages": messages}, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.format_email_list_item", + lambda i, msg, show_unread=True: f"[{i}] {msg['subject']}", + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_inbox + + result = handle_inbox([]) + assert result is True + assert any("Inbox" in p for p in printed) + assert any("[1]" in p for p in printed) + + def test_inbox_resolve_failure(self, monkeypatch): + """When resolve_inbox_target fails, handle_inbox returns False.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.resolve_inbox_target", + lambda first_arg, repo_root, get_branch_fn, get_user_fn: ( + False, + {"error": "Unknown branch: @fake"}, + ), + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_inbox + + result = handle_inbox(["@fake"]) + assert result is False + assert any("Unknown" in e for e in errors) + + +# =========================================================================== +# handle_view +# =========================================================================== + + +class TestHandleView: + """Tests for email.handle_view orchestrator.""" + + def test_view_no_args_shows_usage(self, monkeypatch): + """Calling view with no args prints usage error.""" + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_view + + result = handle_view([]) + assert result is True + assert any("Usage" in e for e in errors) + + def test_view_marks_opened_and_prints(self, tmp_path, monkeypatch): + """View with valid ID marks as opened and prints header + message.""" + email_data = { + "id": "abc123", + "from": "@sender", + "subject": "Test Subject", + "message": "Test body content", + "status": "opened", + } + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.mark_as_opened", + lambda bp, mid: (True, "Marked as opened", email_data), + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.format_email_header", + lambda ed: "FROM: @sender | SUBJECT: Test Subject", + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_view + + result = handle_view(["abc123"]) + assert result is True + assert any("FROM: @sender" in p for p in printed) + assert any("Test body content" in p for p in printed) + + def test_view_mark_opened_failure(self, tmp_path, monkeypatch): + """When mark_as_opened fails, error is printed.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.mark_as_opened", + lambda bp, mid: (False, "Message not found", None), + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_view + + result = handle_view(["missing_id"]) + assert result is True + assert any("not found" in e.lower() for e in errors) + + def test_view_latest_shortcut(self, tmp_path, monkeypatch): + """'latest' arg resolves to most recent message ID.""" + inbox_data = { + "messages": [ + {"id": "old1", "subject": "Old"}, + {"id": "newest", "subject": "Latest"}, + ] + } + email_data = {"id": "newest", "subject": "Latest", "message": "Latest body", "from": "@x"} + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + _write_inbox(tmp_path, messages=inbox_data["messages"]) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.load_inbox", + lambda f: inbox_data, + ) + + opened_ids: list[str] = [] + + def _mock_mark_opened(bp, mid): + """Track which message IDs were passed to mark_as_opened.""" + opened_ids.append(mid) + return (True, "Opened", email_data) + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.mark_as_opened", + _mock_mark_opened, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.format_email_header", + lambda ed: "HEADER", + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_view + + result = handle_view(["latest"]) + assert result is True + assert opened_ids == ["newest"] + + +# =========================================================================== +# handle_close +# =========================================================================== + + +class TestHandleClose: + """Tests for email.handle_close orchestrator.""" + + def test_close_no_args_shows_usage(self, monkeypatch): + """Close with no args prints usage error.""" + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_close + + result = handle_close([]) + assert result is True + assert any("Usage" in e for e in errors) + + def test_close_single_id(self, tmp_path, monkeypatch): + """Close a single message ID via batch_close.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.batch_close", + lambda bp, ids, fn: ([("msg1", True, "Closed msg1")], 1, 0), + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.error", lambda msg: None) + + from aipass.ai_mail.apps.modules.email import handle_close + + result = handle_close(["msg1"]) + assert result is True + assert any("Closed" in p for p in printed) + + def test_close_all(self, tmp_path, monkeypatch): + """Close 'all' delegates to mark_all_read_and_archive.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.mark_all_read_and_archive", + lambda bp: (True, "Archived 5 messages", 5), + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_close + + result = handle_close(["all"]) + assert result is True + assert any("Archived 5" in p for p in printed) + + def test_close_all_failure(self, tmp_path, monkeypatch): + """Close 'all' failure shows error.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.mark_all_read_and_archive", + lambda bp: (False, "Nothing to close", 0), + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_close + + result = handle_close(["all"]) + assert result is True + assert any("Nothing" in e for e in errors) + + def test_close_multiple_ids_triggers_post_ops(self, tmp_path, monkeypatch): + """Batch close of 2+ IDs triggers batch_close_post_ops.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.batch_close", + lambda bp, ids, fn: ( + [("m1", True, "Closed m1"), ("m2", True, "Closed m2")], + 2, + 0, + ), + ) + post_ops_called = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.batch_close_post_ops", + lambda bp, push_fn, central_fn, purge_fn: post_ops_called.append(True), + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.error", lambda msg: None) + + from aipass.ai_mail.apps.modules.email import handle_close + + result = handle_close(["m1", "m2"]) + assert result is True + assert len(post_ops_called) == 1 + + +# =========================================================================== +# handle_reply +# =========================================================================== + + +class TestHandleReply: + """Tests for email.handle_reply orchestrator.""" + + def test_reply_too_few_args(self, monkeypatch): + """Reply with < 2 args prints usage error.""" + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_reply + + result = handle_reply(["only_id"]) + assert result is True + assert any("Usage" in e for e in errors) + + def test_reply_message_not_found(self, tmp_path, monkeypatch): + """Reply to nonexistent message shows error.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_email_by_id", + lambda inbox_file, msg_id: None, + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + _write_inbox(tmp_path) + + from aipass.ai_mail.apps.modules.email import handle_reply + + result = handle_reply(["missing_id", "my reply"]) + assert result is True + assert any("not found" in e.lower() for e in errors) + + def test_reply_success(self, tmp_path, monkeypatch): + """Successful reply prints success message.""" + original = {"id": "msg1", "from": "@sender", "subject": "Re: test"} + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_email_by_id", + lambda inbox_file, msg_id: original, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.send_reply", + lambda bp, orig, msg: (True, "Reply sent to @sender", "reply_001"), + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + _write_inbox(tmp_path) + + from aipass.ai_mail.apps.modules.email import handle_reply + + result = handle_reply(["msg1", "Thanks!"]) + assert result is True + assert any("Reply sent" in p for p in printed) + + def test_reply_send_failure(self, tmp_path, monkeypatch): + """Failed reply shows error message.""" + original = {"id": "msg1", "from": "@sender", "subject": "test"} + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_email_by_id", + lambda inbox_file, msg_id: original, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.send_reply", + lambda bp, orig, msg: (False, "Delivery failed", None), + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + _write_inbox(tmp_path) + + from aipass.ai_mail.apps.modules.email import handle_reply + + result = handle_reply(["msg1", "reply text"]) + assert result is True + assert any("Delivery failed" in e for e in errors) + + +# =========================================================================== +# handle_sent +# =========================================================================== + + +class TestHandleSent: + """Tests for email.handle_sent orchestrator.""" + + def test_sent_no_folder(self, tmp_path, monkeypatch): + """No sent folder prints 'No sent messages'.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_sent + + result = handle_sent([]) + assert result is True + assert any("No sent" in p for p in printed) + + def test_sent_with_files(self, tmp_path, monkeypatch): + """Sent folder with files loads and displays them.""" + sent_folder = tmp_path / ".ai_mail.local" / "sent" + sent_folder.mkdir(parents=True) + email_data = {"id": "s1", "to": "@target", "subject": "Sent test"} + (sent_folder / "email_001.json").write_text(json.dumps(email_data), encoding="utf-8") + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.load_email_file", + lambda f: email_data, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.format_email_list_item", + lambda i, data, show_unread=True: f"[{i}] {data['subject']}", + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_sent + + result = handle_sent([]) + assert result is True + assert any("Sent Messages" in p for p in printed) + assert any("[1]" in p for p in printed) + + def test_sent_empty_folder(self, tmp_path, monkeypatch): + """Sent folder exists but has no JSON files.""" + sent_folder = tmp_path / ".ai_mail.local" / "sent" + sent_folder.mkdir(parents=True) + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email._resolve_branch_path", + lambda: tmp_path, + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_sent + + result = handle_sent([]) + assert result is True + assert any("No sent" in p for p in printed) + + +# =========================================================================== +# handle_contacts +# =========================================================================== + + +class TestHandleContacts: + """Tests for email.handle_contacts orchestrator.""" + + def test_contacts_displays_branches(self, monkeypatch): + """Contacts lists all registered branches.""" + branches = [ + {"email": "@alpha", "name": "ALPHA", "path": "/src/alpha"}, + {"email": "@beta", "name": "BETA", "path": "/src/beta"}, + ] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_all_branches", + lambda: branches, + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_contacts + + result = handle_contacts([]) + assert result is True + assert any("2 branches" in p for p in printed) + # Branches should appear sorted by email + alpha_lines = [p for p in printed if "@alpha" in p] + assert len(alpha_lines) > 0 + + def test_contacts_empty(self, monkeypatch): + """No contacts found prints error.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_all_branches", + lambda: [], + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_contacts + + result = handle_contacts([]) + assert result is True + assert any("No contacts" in e for e in errors) + + def test_contacts_exception_handled(self, monkeypatch): + """Exception in get_all_branches is caught and error is shown.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.get_all_branches", + lambda: (_ for _ in ()).throw(RuntimeError("Registry unavailable")), + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_contacts + + result = handle_contacts([]) + assert result is True + assert any("Registry unavailable" in e for e in errors) + + +# =========================================================================== +# handle_register +# =========================================================================== + + +class TestHandleRegister: + """Tests for email.handle_register orchestrator.""" + + def test_register_too_few_args(self, monkeypatch): + """Register with < 2 args prints usage error.""" + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + + from aipass.ai_mail.apps.modules.email import handle_register + + result = handle_register(["@branch"]) + assert result is True + assert any("Usage" in e for e in errors) + + def test_register_success(self, monkeypatch): + """Successful registration prints green confirmation.""" + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + # register_contact is imported locally inside handle_register, + # so patch at the handler source module + with patch( + "aipass.ai_mail.apps.handlers.email.contacts.register_contact", + return_value=True, + ): + from aipass.ai_mail.apps.modules.email import handle_register + + result = handle_register(["@devpulse", "/path/to/inbox"]) + assert result is True + assert any("Registered" in p and "devpulse" in p for p in printed) + + def test_register_failure(self, monkeypatch): + """Failed registration prints error.""" + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email.error", + lambda msg: errors.append(msg), + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + with patch( + "aipass.ai_mail.apps.handlers.email.contacts.register_contact", + return_value=False, + ): + from aipass.ai_mail.apps.modules.email import handle_register + + result = handle_register(["@badstuff", "/path/to/inbox"]) + assert result is True + assert any("Failed" in e for e in errors) + + def test_register_with_project_arg(self, monkeypatch): + """Third arg is passed as project name.""" + registered_args: list[tuple] = [] + + def _mock_register(name, project, path): + """Capture register_contact arguments for assertion.""" + registered_args.append((name, project, path)) + return True + + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + with patch( + "aipass.ai_mail.apps.handlers.email.contacts.register_contact", + side_effect=_mock_register, + ): + from aipass.ai_mail.apps.modules.email import handle_register + + result = handle_register(["@vera", "/path/to/inbox", "VeraStudio"]) + assert result is True + assert registered_args[0] == ("vera", "VeraStudio", "/path/to/inbox") + + +# =========================================================================== +# handle_send (from email_send.py) +# =========================================================================== + + +class TestHandleSend: + """Tests for email_send.handle_send orchestrator.""" + + def test_send_direct_single_recipient(self, monkeypatch): + """Direct send to a single recipient calls send_to_single.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: { + "mode": "direct", + "recipients": ["@target"], + "subject": "Hello", + "message": "World", + "auto_execute": False, + "reply_to": None, + "no_memory_save": False, + "from_branch": None, + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_dispatch_target", + lambda branch, auto, fn: None, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_sender_info", + lambda fb, rr, amd, gbe, gcu: { + "email_address": "@ai_mail", + "display_name": "AI_MAIL", + "mailbox_path": "/tmp/mailbox", + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.send_to_single", + lambda *a, **kw: (True, None), + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send(["@target", "Hello", "World"]) + assert result is True + assert any("sent" in p.lower() and "@target" in p for p in printed) + + def test_send_error_mode(self, monkeypatch): + """Parse error returns False and prints error.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: { + "mode": "error", + "error": "Usage: send @recipient [subject] [message]", + }, + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.error", + lambda msg: errors.append(msg), + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send(["bad", "args"]) + assert result is False + assert any("Usage" in e for e in errors) + + def test_send_delivery_failure(self, monkeypatch): + """When send_to_single returns failure, error is printed.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: { + "mode": "direct", + "recipients": ["@target"], + "subject": "Sub", + "message": "Msg", + "auto_execute": False, + "reply_to": None, + "no_memory_save": False, + "from_branch": None, + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_dispatch_target", + lambda branch, auto, fn: None, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_sender_info", + lambda fb, rr, amd, gbe, gcu: { + "email_address": "@ai_mail", + "display_name": "AI_MAIL", + "mailbox_path": "/tmp/mailbox", + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.send_to_single", + lambda *a, **kw: (False, "Branch not found"), + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.dispatch_send_error", + lambda *a, **kw: None, + ) + errors: list[str] = [] + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.error", + lambda msg: errors.append(msg), + ) + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send(["@target", "Sub", "Msg"]) + assert result is False + assert any("Branch not found" in e for e in errors) + + def test_send_interactive_mode(self, monkeypatch): + """Interactive mode is triggered when parse returns mode='interactive'.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: {"mode": "interactive"}, + ) + # _send_interactive calls get_all_branches and collect_interactive_input + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.get_all_branches", + lambda: [{"name": "A", "email": "@a"}], + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.collect_interactive_input", + lambda branches: None, # User cancelled + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send([]) + assert result is False + assert any("Cancelled" in p for p in printed) + + def test_send_dispatch_fires_trigger(self, monkeypatch): + """With auto_execute, dispatch trigger is fired after successful send.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: { + "mode": "direct", + "recipients": ["@target"], + "subject": "Dispatch Task", + "message": "Do the thing", + "auto_execute": True, + "reply_to": None, + "no_memory_save": False, + "from_branch": None, + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_dispatch_target", + lambda branch, auto, fn: "@target", + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_sender_info", + lambda fb, rr, amd, gbe, gcu: { + "email_address": "@ai_mail", + "display_name": "AI_MAIL", + "mailbox_path": "/tmp/mailbox", + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.prepend_dispatch_header", + lambda msg, no_memory_save=False: f"[DISPATCH] {msg}", + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.send_to_single", + lambda *a, **kw: (True, None), + ) + + trigger_calls: list[tuple] = [] + mock_trigger = MagicMock() + mock_trigger.fire = lambda event, **kw: trigger_calls.append((event, kw)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.trigger", mock_trigger) + + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: None + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send(["@target", "Dispatch Task", "Do the thing", "--dispatch"]) + assert result is True + assert len(trigger_calls) == 1 + assert trigger_calls[0][0] == "email_dispatched" + assert trigger_calls[0][1]["to"] == "@target" + + def test_send_group_multiple_recipients(self, monkeypatch): + """Group send to multiple recipients calls _send_direct for each.""" + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.parse_send_args", + lambda args: { + "mode": "direct", + "recipients": ["@alpha", "@beta"], + "subject": "Group msg", + "message": "Hi all", + "auto_execute": False, + "reply_to": None, + "no_memory_save": False, + "from_branch": None, + }, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_dispatch_target", + lambda branch, auto, fn: None, + ) + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.resolve_sender_info", + lambda fb, rr, amd, gbe, gcu: { + "email_address": "@ai_mail", + "display_name": "AI_MAIL", + "mailbox_path": "/tmp/mailbox", + }, + ) + + sent_to: list[str] = [] + + def mock_send_single(*args, **kwargs): + """Track which branches receive send_to_single calls.""" + sent_to.append(args[0]) + return (True, None) + + monkeypatch.setattr( + "aipass.ai_mail.apps.modules.email_send.send_to_single", + mock_send_single, + ) + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None) + + from aipass.ai_mail.apps.modules.email_send import handle_send + + result = handle_send(["@alpha", "@beta", "Group msg", "Hi all"]) + assert result is True + # Both recipients should have been sent to + assert "@alpha" in sent_to + assert "@beta" in sent_to + assert any("Group send complete" in p for p in printed) + + +# =========================================================================== +# handle_command dispatch table +# =========================================================================== + + +class TestHandleCommand: + """Tests for the top-level handle_command router.""" + + def test_unknown_command_returns_false(self): + """Unknown command returns False.""" + from aipass.ai_mail.apps.modules.email import handle_command + + result = handle_command("nonexistent", []) + assert result is False + + def test_help_flag_prints_help(self, monkeypatch): + """--help prints help text and returns True.""" + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_command + + result = handle_command("--help", []) + assert result is True + assert any("Email Module" in p for p in printed) + + def test_command_with_help_arg(self, monkeypatch): + """Any valid command with 'help' as first arg prints help.""" + printed: list[str] = [] + mock_console = MagicMock() + mock_console.print = lambda msg, **kw: printed.append(str(msg)) + monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console) + + from aipass.ai_mail.apps.modules.email import handle_command + + result = handle_command("inbox", ["help"]) + assert result is True + assert any("Email Module" in p for p in printed) diff --git a/src/aipass/ai_mail/tests/test_error_dispatch.py b/src/aipass/ai_mail/tests/test_error_dispatch.py new file mode 100644 index 00000000..61be67b9 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_error_dispatch.py @@ -0,0 +1,205 @@ +"""Tests for email error dispatch handler -- error report building, dispatch, and delivery callbacks.""" + +import pytest +from unittest.mock import patch, MagicMock + +from aipass.ai_mail.apps.handlers.email.error_dispatch import ( + build_error_report, + dispatch_send_error, + on_email_delivered, +) + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(): + """Prevent log_operation from writing real JSON files during tests.""" + with patch("aipass.ai_mail.apps.handlers.email.error_dispatch.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +# ---- build_error_report tests -------------------------------- + + +def test_build_error_report_basic_structure(monkeypatch): + """Error report contains all required email fields.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "trigger") + + result = build_error_report("@backup", "Deploy task", "Connection refused") + + assert isinstance(result, dict) + assert result["from"] == "@ai_mail" + assert result["from_name"] == "AI_MAIL" + assert result["to"] == "@drone" + assert result["auto_execute"] is False + assert result["priority"] == "normal" + assert result["reply_to"] == "@devpulse" + assert "timestamp" in result + assert len(result["timestamp"]) == 19 # "YYYY-MM-DD HH:MM:SS" + + +def test_build_error_report_subject_includes_recipient_and_error(monkeypatch): + """Subject line contains the failed recipient and truncated error message.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "trigger") + + result = build_error_report("@backup", "Deploy task", "Connection refused") + + assert "@backup" in result["subject"] + assert "Connection refused" in result["subject"] + assert result["subject"].startswith("[ERROR]") + + +def test_build_error_report_message_body_content(monkeypatch): + """Message body contains sender, recipient, subject, and error details.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow") + + result = build_error_report("@memory", "Vectorize data", "Timeout after 120s") + + body = result["message"] + assert "@flow" in body + assert "@memory" in body + assert "Vectorize data" in body + assert "Timeout after 120s" in body + assert "auto-dispatched" in body + + +def test_build_error_report_with_env_var_set(monkeypatch): + """Uses AIPASS_CALLER_BRANCH env var for sender in the body.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "devpulse") + + result = build_error_report("@flow", "Status check", "Not found") + + assert "@devpulse" in result["message"] + + +def test_build_error_report_without_env_var(monkeypatch): + """Defaults to @ai_mail sender in the body when env var is unset.""" + monkeypatch.delenv("AIPASS_CALLER_BRANCH", raising=False) + + result = build_error_report("@flow", "Status check", "Not found") + + assert "@ai_mail" in result["message"] + + +def test_build_error_report_env_var_with_at_prefix(monkeypatch): + """Handles AIPASS_CALLER_BRANCH that already has @ prefix.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "@trigger") + + result = build_error_report("@backup", "task", "error") + + # Should normalize to @trigger (not @@trigger) + assert "@@" not in result["message"] + assert "@trigger" in result["message"] + + +def test_build_error_report_long_error_truncated_in_subject(monkeypatch): + """Error message in subject is truncated to 50 chars.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "trigger") + long_error = "A" * 100 + + result = build_error_report("@backup", "task", long_error) + + # The subject uses error_msg[:50] + assert len(result["subject"]) < 200 # reasonable length + assert "A" * 50 in result["subject"] + + +# ---- dispatch_send_error tests -------------------------------- + + +def test_dispatch_send_error_success(monkeypatch): + """Returns True when deliver_fn succeeds.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "trigger") + mock_deliver = MagicMock() + + result = dispatch_send_error("@backup", "task", "error msg", mock_deliver) + + assert result is True + mock_deliver.assert_called_once() + args = mock_deliver.call_args[0] + assert args[0] == "@drone" + assert isinstance(args[1], dict) + assert args[1]["to"] == "@drone" + + +def test_dispatch_send_error_failure_returns_false(monkeypatch): + """Returns False when deliver_fn raises an exception.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "trigger") + mock_deliver = MagicMock(side_effect=RuntimeError("network error")) + + result = dispatch_send_error("@backup", "task", "error msg", mock_deliver) + + assert result is False + mock_deliver.assert_called_once() + + +def test_dispatch_send_error_passes_correct_email_data(monkeypatch): + """Verify the email_data dict passed to deliver_fn has expected keys.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow") + captured = {} + + def capture_deliver(target, data): + """Capture deliver_fn arguments for assertion.""" + captured.update(data) + + dispatch_send_error("@memory", "Vectorize", "Timeout", capture_deliver) + + assert captured["from"] == "@ai_mail" + assert captured["to"] == "@drone" + assert "@memory" in captured["subject"] + + +# ---- on_email_delivered tests -------------------------------- + + +def test_on_email_delivered_with_both_callbacks(): + """Both callbacks are invoked when provided.""" + push_fn = MagicMock() + update_fn = MagicMock() + branch_path = "/some/path" + + on_email_delivered(branch_path, 3, 1, 10, push_fn, update_fn) + + push_fn.assert_called_once_with(branch_path) + update_fn.assert_called_once_with() + + +def test_on_email_delivered_with_none_callbacks(): + """No error when both callbacks are None.""" + on_email_delivered("/some/path", 3, 1, 10, None, None) + + +def test_on_email_delivered_dashboard_failure_does_not_block_central(): + """Dashboard failure does not prevent central update from running.""" + push_fn = MagicMock(side_effect=RuntimeError("dashboard broken")) + update_fn = MagicMock() + + on_email_delivered("/some/path", 3, 1, 10, push_fn, update_fn) + + push_fn.assert_called_once() + update_fn.assert_called_once() + + +def test_on_email_delivered_central_failure_does_not_raise(): + """Central update failure is caught silently.""" + push_fn = MagicMock() + update_fn = MagicMock(side_effect=RuntimeError("central broken")) + + on_email_delivered("/some/path", 3, 1, 10, push_fn, update_fn) + + push_fn.assert_called_once() + update_fn.assert_called_once() + + +def test_on_email_delivered_both_fail_no_exception(): + """Both callbacks failing does not raise any exception.""" + push_fn = MagicMock(side_effect=RuntimeError("push fail")) + update_fn = MagicMock(side_effect=RuntimeError("update fail")) + + on_email_delivered("/some/path", 3, 1, 10, push_fn, update_fn) + + push_fn.assert_called_once() + update_fn.assert_called_once() diff --git a/src/aipass/ai_mail/tests/test_footer.py b/src/aipass/ai_mail/tests/test_footer.py new file mode 100644 index 00000000..d47c5534 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_footer.py @@ -0,0 +1,88 @@ +# =================== AIPass ==================== +# Name: test_footer.py +# Description: Tests for email footer handler +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for email footer handler -- get_footer, append_footer.""" + +import pytest +from unittest.mock import MagicMock + +import aipass.ai_mail.apps.handlers.email.footer as mod + + +# --- Fixtures -------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _suppress_log_operation(monkeypatch): + """Prevent json_handler.log_operation from touching real files.""" + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +# --- get_footer tests ------------------------------------------------ + + +def test_get_footer_returns_string(): + """get_footer returns a string.""" + result = mod.get_footer() + assert isinstance(result, str) + + +def test_get_footer_matches_constant(): + """get_footer returns the STANDARD_FOOTER constant.""" + result = mod.get_footer() + assert result == mod.STANDARD_FOOTER + + +def test_get_footer_contains_checklist(): + """Footer contains the task checklist markers.""" + result = mod.get_footer() + assert "TASK CHECKLIST" in result + assert "SEEDGO CHECK" in result + assert "UPDATE MEMORIES" in result + assert "CLOSE FPLAN" in result + assert "EMAIL SENDER" in result + + +# --- append_footer tests --------------------------------------------- + + +def test_append_footer_appends_to_message(): + """append_footer adds the standard footer to the end of a message.""" + message = "Hello, this is the task body." + result = mod.append_footer(message) + assert result.startswith(message) + assert result.endswith(mod.STANDARD_FOOTER) + + +def test_append_footer_preserves_original_message(): + """Original message text is intact in the result.""" + message = "Complete the integration for module X." + result = mod.append_footer(message) + assert message in result + + +def test_append_footer_logs_operation(_suppress_log_operation: MagicMock): + """append_footer calls json_handler.log_operation with message length.""" + message = "Test message body" + mod.append_footer(message) + _suppress_log_operation.log_operation.assert_called_once_with("append_footer", {"message_length": len(message)}) + + +def test_append_footer_empty_message(): + """append_footer works with an empty message string.""" + result = mod.append_footer("") + assert result == mod.STANDARD_FOOTER + + +def test_append_footer_multiline_message(): + """append_footer handles multi-line messages correctly.""" + message = "Line 1\nLine 2\nLine 3" + result = mod.append_footer(message) + assert result == message + mod.STANDARD_FOOTER diff --git a/src/aipass/ai_mail/tests/test_format.py b/src/aipass/ai_mail/tests/test_format.py new file mode 100644 index 00000000..60eb507b --- /dev/null +++ b/src/aipass/ai_mail/tests/test_format.py @@ -0,0 +1,284 @@ +"""Tests for email formatting handler -- lookup, preview, header, list item.""" + +import json + +import pytest +from unittest.mock import MagicMock +from pathlib import Path + +import aipass.ai_mail.apps.handlers.email.format as mod + + +# --- Fixtures -------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _suppress_log_operation(monkeypatch): + """Prevent json_handler.log_operation from touching real files.""" + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +@pytest.fixture() +def registry_file(tmp_path, monkeypatch): + """Create a temporary AIPASS_REGISTRY.json and patch REGISTRY_PATH.""" + registry_data = { + "branches": [ + {"name": "TEAM_1", "alias": "Team Alpha"}, + {"name": "VERA", "alias": "Vera"}, + {"name": "NO_ALIAS", "alias": ""}, + {"name": "NULL_ALIAS"}, + ] + } + reg_path = tmp_path / "AIPASS_REGISTRY.json" + reg_path.write_text(json.dumps(registry_data), encoding="utf-8") + monkeypatch.setattr(mod, "REGISTRY_PATH", reg_path) + return reg_path + + +# --- lookup_branch_alias tests --------------------------------------- + + +def test_lookup_branch_alias_returns_alias(registry_file): + """Returns alias when branch has a non-empty alias.""" + result = mod.lookup_branch_alias("TEAM_1") + assert result == "Team Alpha" + + +def test_lookup_branch_alias_empty_alias_returns_none(registry_file): + """Returns None when branch has an empty string alias.""" + result = mod.lookup_branch_alias("NO_ALIAS") + assert result is None + + +def test_lookup_branch_alias_missing_alias_key_returns_none(registry_file): + """Returns None when branch dict has no 'alias' key.""" + result = mod.lookup_branch_alias("NULL_ALIAS") + assert result is None + + +def test_lookup_branch_alias_unknown_branch_returns_none(registry_file): + """Returns None for a branch not in the registry.""" + result = mod.lookup_branch_alias("NONEXISTENT") + assert result is None + + +def test_lookup_branch_alias_file_missing_returns_none(tmp_path, monkeypatch): + """Returns None when REGISTRY_PATH points to a non-existent file.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", tmp_path / "missing.json") + result = mod.lookup_branch_alias("TEAM_1") + assert result is None + + +def test_lookup_branch_alias_invalid_json_returns_none(tmp_path, monkeypatch): + """Returns None when REGISTRY_PATH contains invalid JSON.""" + bad_file = tmp_path / "bad.json" + bad_file.write_text("not valid json {{{", encoding="utf-8") + monkeypatch.setattr(mod, "REGISTRY_PATH", bad_file) + result = mod.lookup_branch_alias("TEAM_1") + assert result is None + + +# --- format_sender_display tests ------------------------------------- + + +def test_format_sender_display_with_alias(registry_file): + """Uses alias when branch has one.""" + result = mod.format_sender_display("TEAM_1", "@team_1") + assert result == "Team Alpha (@team_1)" + + +def test_format_sender_display_without_alias(registry_file): + """Falls back to from_name when branch has no alias.""" + result = mod.format_sender_display("NO_ALIAS", "@no_alias") + assert result == "NO_ALIAS (@no_alias)" + + +def test_format_sender_display_unknown_branch(registry_file): + """Falls back to from_name for unknown branch.""" + result = mod.format_sender_display("UNKNOWN", "@unknown") + assert result == "UNKNOWN (@unknown)" + + +# --- format_email_preview tests -------------------------------------- + + +def test_format_email_preview_short_message(): + """Returns full message when under max_length.""" + msg = "Short message" + result = mod.format_email_preview(msg, max_length=100) + assert result == msg + + +def test_format_email_preview_exact_length(): + """Returns full message when exactly at max_length.""" + msg = "x" * 100 + result = mod.format_email_preview(msg, max_length=100) + assert result == msg + assert "..." not in result + + +def test_format_email_preview_truncates_long_message(): + """Truncates and adds ellipsis when over max_length.""" + msg = "a" * 150 + result = mod.format_email_preview(msg, max_length=100) + assert len(result) == 103 # 100 chars + "..." + assert result.endswith("...") + + +def test_format_email_preview_empty_message(): + """Returns empty string for empty message.""" + result = mod.format_email_preview("") + assert result == "" + + +def test_format_email_preview_default_max_length(): + """Default max_length is 100.""" + msg = "b" * 101 + result = mod.format_email_preview(msg) + assert result == "b" * 100 + "..." + + +# --- format_email_header tests --------------------------------------- + + +def test_format_email_header_contains_all_fields(monkeypatch): + """Header includes From, Date, Subject, and separator lines.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "from_name": "TEAM_1", + "from": "@team_1", + "timestamp": "2026-04-25T10:00:00", + "subject": "Test Subject", + } + result = mod.format_email_header(email_data) + assert "From: TEAM_1 (@team_1)" in result + assert "Date: 2026-04-25T10:00:00" in result + assert "Subject: Test Subject" in result + assert "=" * 70 in result + + +def test_format_email_header_missing_fields(monkeypatch): + """Uses defaults for missing email_data fields.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + result = mod.format_email_header({}) + assert "From: Unknown (unknown)" in result + assert "Date: Unknown" in result + assert "Subject: No Subject" in result + + +def test_format_email_header_logs_operation( + _suppress_log_operation: MagicMock, + monkeypatch, +): + """format_email_header calls json_handler.log_operation.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data: dict[str, str] = {"subject": "Log Test"} + mod.format_email_header(email_data) + _suppress_log_operation.log_operation.assert_called_once_with("format_email_header", {"subject": "Log Test"}) + + +def test_format_email_header_with_alias(registry_file): + """Header uses alias when branch has one in the registry.""" + email_data = { + "from_name": "VERA", + "from": "@vera", + "timestamp": "2026-04-25", + "subject": "Alias Test", + } + result = mod.format_email_header(email_data) + assert "From: Vera (@vera)" in result + + +# --- format_email_list_item tests ------------------------------------ + + +def test_format_email_list_item_new_message(monkeypatch): + """New/unread message shows the new-mail emoji marker.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "id": "abc123", + "from_name": "SENDER", + "from": "@sender", + "timestamp": "2026-04-25", + "subject": "New Mail", + "message": "Hello world", + "status": "new", + } + result = mod.format_email_list_item(1, email_data) + assert "\U0001f4e8" in result # new-mail emoji + assert "[abc123]" in result + assert "Subject: New Mail" in result + + +def test_format_email_list_item_opened_message(monkeypatch): + """Opened message shows the opened-mailbox emoji marker.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "id": "def456", + "from_name": "SENDER", + "from": "@sender", + "timestamp": "2026-04-25", + "subject": "Read Mail", + "message": "Already read", + "status": "opened", + } + result = mod.format_email_list_item(1, email_data) + assert "\U0001f4ec" in result # opened-mailbox emoji + + +def test_format_email_list_item_read_fallback(monkeypatch): + """Falls back to 'read' field when 'status' is absent.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "id": "ghi789", + "from_name": "SENDER", + "from": "@sender", + "timestamp": "2026-04-25", + "subject": "Legacy Mail", + "message": "Old format", + "read": True, + } + result = mod.format_email_list_item(1, email_data) + assert "\U0001f4ec" in result # opened-mailbox emoji (read=True) + + +def test_format_email_list_item_show_unread_false(monkeypatch): + """When show_unread=False, shows 'To:' instead of sender with emoji.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "id": "jkl012", + "to": "@recipient", + "timestamp": "2026-04-25", + "subject": "Sent Mail", + "message": "Outgoing message", + } + result = mod.format_email_list_item(1, email_data, show_unread=False) + assert "To: @recipient" in result + assert "\U0001f4e8" not in result + assert "\U0001f4ec" not in result + + +def test_format_email_list_item_missing_fields(monkeypatch): + """Uses defaults for missing email_data fields.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + result = mod.format_email_list_item(1, {}) + assert "[????????]" in result + assert "Subject: No Subject" in result + + +def test_format_email_list_item_truncates_long_message(monkeypatch): + """Long message is truncated in the preview.""" + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/nonexistent")) + email_data = { + "id": "trunc1", + "from_name": "SENDER", + "from": "@sender", + "timestamp": "2026-04-25", + "subject": "Long Body", + "message": "z" * 200, + "status": "new", + } + result = mod.format_email_list_item(1, email_data) + assert "..." in result diff --git a/src/aipass/ai_mail/tests/test_header.py b/src/aipass/ai_mail/tests/test_header.py new file mode 100644 index 00000000..63f92cfc --- /dev/null +++ b/src/aipass/ai_mail/tests/test_header.py @@ -0,0 +1,114 @@ +"""Tests for email header handler -- get_dispatch_header, prepend_dispatch_header.""" + +import pytest +from unittest.mock import MagicMock + +import aipass.ai_mail.apps.handlers.email.header as mod + + +# --- Fixtures -------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _suppress_log_operation(monkeypatch): + """Prevent json_handler.log_operation from touching real files.""" + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +# --- get_dispatch_header tests --------------------------------------- + + +def test_get_dispatch_header_default_returns_standard(): + """Default call returns DISPATCH_HEADER.""" + result = mod.get_dispatch_header() + assert result == mod.DISPATCH_HEADER + + +def test_get_dispatch_header_no_memory_save_returns_variant(): + """no_memory_save=True returns NO_MEMORY_SAVE_HEADER.""" + result = mod.get_dispatch_header(no_memory_save=True) + assert result == mod.NO_MEMORY_SAVE_HEADER + + +def test_get_dispatch_header_false_returns_standard(): + """Explicit no_memory_save=False returns DISPATCH_HEADER.""" + result = mod.get_dispatch_header(no_memory_save=False) + assert result == mod.DISPATCH_HEADER + + +def test_dispatch_header_contains_memory_reminder(): + """Standard header reminds agents to update memories.""" + result = mod.get_dispatch_header() + assert "UPDATE YOUR MEMORIES" in result + assert "NOT optional" in result + + +def test_no_memory_save_header_contains_optional_directive(): + """No-memory-save header marks memory update as OPTIONAL.""" + result = mod.get_dispatch_header(no_memory_save=True) + assert "OPTIONAL" in result + assert "Do NOT log this task" in result + + +def test_headers_are_distinct(): + """The two header variants are different strings.""" + standard = mod.get_dispatch_header(no_memory_save=False) + no_save = mod.get_dispatch_header(no_memory_save=True) + assert standard != no_save + + +# --- prepend_dispatch_header tests ----------------------------------- + + +def test_prepend_dispatch_header_default(): + """Prepends standard dispatch header to message.""" + message = "Please complete task X." + result = mod.prepend_dispatch_header(message) + assert result.startswith(mod.DISPATCH_HEADER) + assert result.endswith(message) + + +def test_prepend_dispatch_header_no_memory_save(): + """Prepends no-memory-save header when flag is set.""" + message = "Private task." + result = mod.prepend_dispatch_header(message, no_memory_save=True) + assert result.startswith(mod.NO_MEMORY_SAVE_HEADER) + assert result.endswith(message) + + +def test_prepend_dispatch_header_preserves_message(): + """Original message text is fully preserved in result.""" + message = "Multi\nline\nmessage\nbody" + result = mod.prepend_dispatch_header(message) + assert message in result + + +def test_prepend_dispatch_header_logs_operation( + _suppress_log_operation: MagicMock, +): + """prepend_dispatch_header calls json_handler.log_operation.""" + mod.prepend_dispatch_header("test", no_memory_save=False) + _suppress_log_operation.log_operation.assert_called_once_with("prepend_dispatch_header", {"no_memory_save": False}) + + +def test_prepend_dispatch_header_logs_no_memory_save_flag( + _suppress_log_operation: MagicMock, +): + """Log call captures no_memory_save=True when set.""" + mod.prepend_dispatch_header("test", no_memory_save=True) + _suppress_log_operation.log_operation.assert_called_once_with("prepend_dispatch_header", {"no_memory_save": True}) + + +def test_prepend_dispatch_header_empty_message(): + """Works with an empty message string.""" + result = mod.prepend_dispatch_header("") + assert result == mod.DISPATCH_HEADER + + +def test_prepend_dispatch_header_result_is_header_plus_message(): + """Result is exactly header concatenated with message.""" + message = "Exact concatenation test." + result = mod.prepend_dispatch_header(message, no_memory_save=False) + assert result == mod.DISPATCH_HEADER + message diff --git a/src/aipass/ai_mail/tests/test_inbox_cleanup.py b/src/aipass/ai_mail/tests/test_inbox_cleanup.py new file mode 100644 index 00000000..11e80bde --- /dev/null +++ b/src/aipass/ai_mail/tests/test_inbox_cleanup.py @@ -0,0 +1,315 @@ +# =================== AIPass ==================== +# Name: test_inbox_cleanup.py +# Description: Tests for inbox cleanup handler +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for inbox cleanup handler -- mark_all_read, mark_as_opened, mark_as_closed.""" + +import json +from contextlib import contextmanager + +import pytest +from pathlib import Path +from unittest.mock import MagicMock + +import aipass.ai_mail.apps.handlers.email.inbox_cleanup as mod + + +# ---- Fixtures ------------------------------------------------ + + +@contextmanager +def _noop_lock(_path: Path): + """Dummy context manager replacing the real inbox file lock.""" + yield + + +@pytest.fixture(autouse=True) +def _silence_json_handler(monkeypatch): + """Prevent log_operation from writing real JSON files during tests.""" + mock_jh = MagicMock() + mock_jh.log_operation.return_value = True + monkeypatch.setattr(mod, "json_handler", mock_jh) + return mock_jh + + +@pytest.fixture(autouse=True) +def _mock_inbox_lock(monkeypatch): + """Replace _get_inbox_lock so it returns a no-op context manager.""" + monkeypatch.setattr(mod, "_get_inbox_lock", lambda: _noop_lock) + + +@pytest.fixture(autouse=True) +def _mock_dashboard(monkeypatch): + """Replace _get_push_dashboard_update with a no-op.""" + monkeypatch.setattr(mod, "_get_push_dashboard_update", lambda: lambda _bp: None) + + +@pytest.fixture(autouse=True) +def _mock_central(monkeypatch): + """Replace _get_update_central with a no-op.""" + monkeypatch.setattr(mod, "_get_update_central", lambda: lambda: None) + + +@pytest.fixture(autouse=True) +def _mock_deleted_purge(monkeypatch): + """Replace _trigger_deleted_purge with a no-op.""" + monkeypatch.setattr(mod, "_trigger_deleted_purge", lambda _bp: None) + + +def _make_inbox(branch_path: Path, messages: list) -> Path: + """Create a branch with .ai_mail.local/inbox.json containing messages.""" + mailbox = branch_path / ".ai_mail.local" + mailbox.mkdir(parents=True, exist_ok=True) + inbox_file = mailbox / "inbox.json" + unread = sum( + 1 for m in messages if m.get("status") == "new" or (m.get("status") is None and not m.get("read", False)) + ) + data = { + "mailbox": "inbox", + "total_messages": len(messages), + "unread_count": unread, + "messages": messages, + } + inbox_file.write_text(json.dumps(data), encoding="utf-8") + return inbox_file + + +# ---- mark_all_read_and_archive tests --------------------------- + + +def test_mark_all_read_and_archive_success(tmp_path: Path): + """Archives all messages and clears inbox.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [ + {"id": "m1", "status": "new", "subject": "First", "read": False}, + {"id": "m2", "status": "opened", "subject": "Second", "read": True}, + ], + ) + + success, message, count = mod.mark_all_read_and_archive(branch_path) + + assert success is True + assert count == 2 + assert "Archived 2" in message + + # Inbox should be empty + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert inbox_data["messages"] == [] + assert inbox_data["total_messages"] == 0 + assert inbox_data["unread_count"] == 0 + + # Deleted folder should have files + deleted_dir = branch_path / ".ai_mail.local" / "deleted" + assert deleted_dir.is_dir() + deleted_files = list(deleted_dir.glob("*.json")) + assert len(deleted_files) == 2 + + +def test_mark_all_read_and_archive_empty_inbox(tmp_path: Path): + """Empty inbox returns success with count 0.""" + branch_path = tmp_path / "branch" + _make_inbox(branch_path, []) + + success, message, count = mod.mark_all_read_and_archive(branch_path) + + assert success is True + assert count == 0 + assert "empty" in message.lower() + + +def test_mark_all_read_and_archive_no_inbox(tmp_path: Path): + """Missing inbox file returns failure.""" + branch_path = tmp_path / "branch" + branch_path.mkdir(parents=True) + + success, message, count = mod.mark_all_read_and_archive(branch_path) + + assert success is False + assert count == 0 + assert "not found" in message.lower() + + +# ---- mark_as_opened tests ------------------------------------- + + +def test_mark_as_opened_success(tmp_path: Path): + """Marks a message as opened and sets read=True.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [ + {"id": "m1", "status": "new", "subject": "Hello", "read": False}, + ], + ) + + success, message, email_data = mod.mark_as_opened(branch_path, "m1") + + assert success is True + assert "opened" in message.lower() + assert email_data is not None + assert email_data["status"] == "opened" + assert email_data["read"] is True + + # Verify inbox file updated + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert inbox_data["messages"][0]["status"] == "opened" + assert inbox_data["unread_count"] == 0 + + +def test_mark_as_opened_not_found(tmp_path: Path): + """Nonexistent message ID returns failure.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [{"id": "m1", "status": "new", "subject": "Hello"}], + ) + + success, message, email_data = mod.mark_as_opened(branch_path, "nonexistent") + + assert success is False + assert "not found" in message.lower() + assert email_data is None + + +def test_mark_as_opened_no_inbox(tmp_path: Path): + """Missing inbox file returns failure.""" + branch_path = tmp_path / "branch" + branch_path.mkdir(parents=True) + + success, message, email_data = mod.mark_as_opened(branch_path, "m1") + + assert success is False + assert "not found" in message.lower() + assert email_data is None + + +def test_mark_as_opened_updates_unread_count(tmp_path: Path): + """Opening one of two new messages reduces unread_count by 1.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [ + {"id": "m1", "status": "new", "subject": "A", "read": False}, + {"id": "m2", "status": "new", "subject": "B", "read": False}, + ], + ) + + mod.mark_as_opened(branch_path, "m1") + + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert inbox_data["unread_count"] == 1 + + +# ---- mark_as_closed_and_archive tests -------------------------- + + +def test_mark_as_closed_and_archive_success(tmp_path: Path): + """Closes message, removes from inbox, saves to deleted/.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [ + {"id": "m1", "status": "opened", "subject": "Task", "read": True}, + {"id": "m2", "status": "new", "subject": "Other", "read": False}, + ], + ) + + success, message = mod.mark_as_closed_and_archive(branch_path, "m1") + + assert success is True + assert "closed" in message.lower() + + # Inbox should have only m2 + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert len(inbox_data["messages"]) == 1 + assert inbox_data["messages"][0]["id"] == "m2" + assert inbox_data["total_messages"] == 1 + + # Deleted folder should have the archived message + deleted_dir = branch_path / ".ai_mail.local" / "deleted" + assert deleted_dir.is_dir() + deleted_files = list(deleted_dir.glob("*.json")) + assert len(deleted_files) == 1 + + +def test_mark_as_closed_and_archive_not_found(tmp_path: Path): + """Nonexistent message ID returns failure.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [{"id": "m1", "status": "new", "subject": "Hello"}], + ) + + success, message = mod.mark_as_closed_and_archive(branch_path, "nonexistent") + + assert success is False + assert "not found" in message.lower() + + +def test_mark_as_closed_and_archive_no_inbox(tmp_path: Path): + """Missing inbox file returns failure.""" + branch_path = tmp_path / "branch" + branch_path.mkdir(parents=True) + + success, message = mod.mark_as_closed_and_archive(branch_path, "m1") + + assert success is False + assert "not found" in message.lower() + + +def test_mark_as_closed_and_archive_skip_post_ops(tmp_path: Path): + """With skip_post_ops=True, message is still archived but post-ops skipped.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [{"id": "m1", "status": "opened", "subject": "Task", "read": True}], + ) + + success, message = mod.mark_as_closed_and_archive(branch_path, "m1", skip_post_ops=True) + + assert success is True + + # Inbox should be empty + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert len(inbox_data["messages"]) == 0 + + # Deleted folder should still have the archived file + deleted_dir = branch_path / ".ai_mail.local" / "deleted" + deleted_files = list(deleted_dir.glob("*.json")) + assert len(deleted_files) == 1 + + +def test_mark_as_closed_and_archive_updates_counts(tmp_path: Path): + """Closing a message updates total_messages and unread_count.""" + branch_path = tmp_path / "branch" + _make_inbox( + branch_path, + [ + {"id": "m1", "status": "new", "subject": "A", "read": False}, + {"id": "m2", "status": "new", "subject": "B", "read": False}, + ], + ) + + mod.mark_as_closed_and_archive(branch_path, "m1") + + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + with open(inbox_file, "r", encoding="utf-8") as f: + inbox_data = json.load(f) + assert inbox_data["total_messages"] == 1 + assert inbox_data["unread_count"] == 1 diff --git a/src/aipass/ai_mail/tests/test_misc_handlers.py b/src/aipass/ai_mail/tests/test_misc_handlers.py new file mode 100644 index 00000000..53392d7c --- /dev/null +++ b/src/aipass/ai_mail/tests/test_misc_handlers.py @@ -0,0 +1,536 @@ +"""Tests for miscellaneous handlers -- central_writer.update_central, dispatch status.check_pid_status, +daemon.run_daemon, json_handler.increment_counter/update_data_metrics, delivery.deliver_to_inbox_file, +dashboard_sync.push_dashboard_update, inbox_resolve.resolve_inbox_target.""" + +import json +import os +import subprocess +import pytest +from pathlib import Path +from unittest.mock import patch, MagicMock + +import aipass.ai_mail.apps.handlers.central_writer as central_mod +import aipass.ai_mail.apps.handlers.dispatch.daemon as daemon_mod +import aipass.ai_mail.apps.handlers.json_utils.json_handler as json_handler_mod +import aipass.ai_mail.apps.handlers.email.delivery as delivery_mod +import aipass.ai_mail.apps.handlers.email.dashboard_sync as dashboard_mod +from aipass.ai_mail.apps.handlers.central_writer import update_central +from aipass.ai_mail.apps.handlers.dispatch.status import check_pid_status +from aipass.ai_mail.apps.handlers.json_utils.json_handler import ( + increment_counter, + update_data_metrics, +) +from aipass.ai_mail.apps.handlers.email.delivery import deliver_to_inbox_file +from aipass.ai_mail.apps.handlers.email.dashboard_sync import push_dashboard_update +from aipass.ai_mail.apps.handlers.email.inbox_resolve import resolve_inbox_target + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler_central(): + """Prevent log_operation in central_writer from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.central_writer.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_json_handler_status(): + """Prevent log_operation in dispatch status from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.dispatch.status.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_json_handler_daemon(): + """Prevent log_operation in daemon from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.dispatch.daemon.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_json_handler_delivery(): + """Prevent log_operation in delivery from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.email.delivery.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_json_handler_dashboard(): + """Prevent log_operation in dashboard_sync from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.email.dashboard_sync.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_json_handler_inbox_resolve(): + """Prevent log_operation in inbox_resolve from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.email.inbox_resolve.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +# ============================================================== +# update_central tests +# ============================================================== + + +def test_update_central_calls_build_and_write(): + """update_central calls aggregate_branch_stats, build_central_data, and write_central_file.""" + mock_stats = {"FLOW": {"unread": 2, "total": 5}} + mock_data = {"service": "ai_mail", "branch_stats": mock_stats} + + with ( + patch.object(central_mod, "aggregate_branch_stats", return_value=mock_stats), + patch.object(central_mod, "build_central_data", return_value=mock_data), + patch.object(central_mod, "write_central_file") as mock_write, + ): + result = update_central() + + assert result == mock_data + mock_write.assert_called_once_with(mock_data) + + +def test_update_central_propagates_error(): + """update_central propagates exceptions from aggregate_branch_stats.""" + with ( + patch.object(central_mod, "aggregate_branch_stats", side_effect=RuntimeError("scan failed")), + pytest.raises(RuntimeError, match="scan failed"), + ): + update_central() + + +# ============================================================== +# check_pid_status tests +# ============================================================== + + +def test_check_pid_status_running(): + """Returns RUNNING for the current process PID.""" + result = check_pid_status(os.getpid()) + + assert result == "RUNNING" + + +def test_check_pid_status_completed(): + """Returns COMPLETED for a dead/nonexistent PID.""" + # Use a PID that almost certainly does not exist + with patch("aipass.ai_mail.apps.handlers.dispatch.status.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=1) + result = check_pid_status(999999999) + + assert result == "COMPLETED" + + +def test_check_pid_status_unknown_on_error(): + """Returns UNKNOWN when subprocess raises an error.""" + with patch("aipass.ai_mail.apps.handlers.dispatch.status.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.SubprocessError("ps failed") + result = check_pid_status(12345) + + assert result == "UNKNOWN" + + +# ============================================================== +# run_daemon tests (minimal -- max_cycles not available, test poll_cycle call) +# ============================================================== + + +def test_daemon_poll_cycle_is_called(tmp_path, monkeypatch): + """run_daemon calls poll_cycle and save_daemon_state in the loop. + + We mock the key dependencies and set SHUTDOWN to True after one cycle + to verify the loop structure works. + """ + monkeypatch.setattr(daemon_mod, "DAEMON_PID_FILE", tmp_path / "daemon.pid") + monkeypatch.setattr(daemon_mod, "CONFIG_FILE", tmp_path / "safety_config.json") + monkeypatch.setattr(daemon_mod, "DAEMON_STATE_FILE", tmp_path / "daemon_state.json") + + poll_calls = [] + + def mock_poll_cycle(config, state): + """Track poll_cycle invocations and trigger shutdown.""" + poll_calls.append(True) + daemon_mod.SHUTDOWN = True + return 0 + + with ( + patch.object(daemon_mod, "_write_pid_file", return_value=True), + patch.object(daemon_mod, "_remove_pid_file"), + patch.object(daemon_mod, "_notify_telegram", return_value=False), + patch.object(daemon_mod, "poll_cycle", side_effect=mock_poll_cycle), + patch.object(daemon_mod, "save_daemon_state"), + patch.object(daemon_mod, "is_kill_switch_active", return_value=False), + patch("os.waitpid", side_effect=ChildProcessError), + ): + # Reset SHUTDOWN before running + daemon_mod.SHUTDOWN = False + daemon_mod.run_daemon() + + assert len(poll_calls) == 1 + + # Clean up global state + daemon_mod.SHUTDOWN = False + + +def test_daemon_exits_if_pid_file_blocked(tmp_path, monkeypatch): + """run_daemon returns immediately when _write_pid_file returns False.""" + monkeypatch.setattr(daemon_mod, "DAEMON_PID_FILE", tmp_path / "daemon.pid") + + with ( + patch.object(daemon_mod, "_write_pid_file", return_value=False), + patch.object(daemon_mod, "poll_cycle") as mock_poll, + ): + daemon_mod.run_daemon() + + mock_poll.assert_not_called() + + +# ============================================================== +# increment_counter tests +# ============================================================== + + +def test_increment_counter_basic(monkeypatch): + """increment_counter loads data, increments, and saves.""" + existing_data = {"created": "2026-01-01", "last_updated": "2026-01-01", "send_count": 5} + + monkeypatch.setattr(json_handler_mod, "ensure_module_jsons", lambda m: True) + monkeypatch.setattr(json_handler_mod, "load_json", lambda m, t: existing_data.copy()) + + saved = {} + + def mock_save(module, json_type, data): + """Capture saved data for assertion.""" + saved.update(data) + return True + + monkeypatch.setattr(json_handler_mod, "save_json", mock_save) + + result = increment_counter("ai_mail", "send_count", 1) + + assert result is True + assert saved["send_count"] == 6 + + +def test_increment_counter_creates_key(monkeypatch): + """increment_counter creates the counter key if it does not exist.""" + existing_data = {"created": "2026-01-01", "last_updated": "2026-01-01"} + + monkeypatch.setattr(json_handler_mod, "ensure_module_jsons", lambda m: True) + monkeypatch.setattr(json_handler_mod, "load_json", lambda m, t: existing_data.copy()) + + saved = {} + + def mock_save(module, json_type, data): + """Capture saved data for assertion.""" + saved.update(data) + return True + + monkeypatch.setattr(json_handler_mod, "save_json", mock_save) + + result = increment_counter("ai_mail", "new_counter", 3) + + assert result is True + assert saved["new_counter"] == 3 + + +def test_increment_counter_returns_false_on_no_data(monkeypatch): + """increment_counter returns False when load_json returns None.""" + monkeypatch.setattr(json_handler_mod, "ensure_module_jsons", lambda m: True) + monkeypatch.setattr(json_handler_mod, "load_json", lambda m, t: None) + + result = increment_counter("ai_mail", "counter") + + assert result is False + + +# ============================================================== +# update_data_metrics tests +# ============================================================== + + +def test_update_data_metrics_basic(monkeypatch): + """update_data_metrics updates multiple keys in data.""" + existing_data = {"created": "2026-01-01", "last_updated": "2026-01-01", "old_key": "old_val"} + + monkeypatch.setattr(json_handler_mod, "ensure_module_jsons", lambda m: True) + monkeypatch.setattr(json_handler_mod, "load_json", lambda m, t: existing_data.copy()) + + saved = {} + + def mock_save(module, json_type, data): + """Capture saved data for assertion.""" + saved.update(data) + return True + + monkeypatch.setattr(json_handler_mod, "save_json", mock_save) + + result = update_data_metrics("ai_mail", status="healthy", uptime=3600) + + assert result is True + assert saved["status"] == "healthy" + assert saved["uptime"] == 3600 + assert saved["old_key"] == "old_val" + + +def test_update_data_metrics_returns_false_on_no_data(monkeypatch): + """update_data_metrics returns False when load_json returns None.""" + monkeypatch.setattr(json_handler_mod, "ensure_module_jsons", lambda m: True) + monkeypatch.setattr(json_handler_mod, "load_json", lambda m, t: None) + + result = update_data_metrics("ai_mail", key="value") + + assert result is False + + +# ============================================================== +# deliver_to_inbox_file tests +# ============================================================== + + +@pytest.fixture +def _noop_inbox_lock(monkeypatch): + """Replace _get_inbox_lock with a no-op context manager.""" + from contextlib import contextmanager + + @contextmanager + def _noop_lock(path): + yield + + monkeypatch.setattr(delivery_mod, "_get_inbox_lock", lambda: _noop_lock) + + +@pytest.fixture(autouse=True) +def _silence_delivery_notifications(): + """Prevent desktop notifications during delivery tests.""" + with patch.object(delivery_mod, "_send_desktop_notification"): + yield + + +def test_deliver_to_inbox_file_happy_path(tmp_path, _noop_inbox_lock): + """Successful delivery writes message to inbox and returns (True, '', reply_id).""" + inbox_file = tmp_path / "inbox.json" + inbox_data = { + "mailbox": "inbox", + "total_messages": 0, + "unread_count": 0, + "messages": [], + } + inbox_file.write_text(json.dumps(inbox_data, indent=2), encoding="utf-8") + + email_data = { + "from": "@sender", + "to": "@target", + "subject": "Test delivery", + "message": "Test body", + "timestamp": "2026-04-01 10:00:00", + "status": "new", + } + + success, error, reply_id = deliver_to_inbox_file(inbox_file, email_data) + + assert success is True + assert error == "" + assert reply_id != "" + assert len(reply_id) == 8 + + with open(inbox_file, "r", encoding="utf-8") as f: + result = json.load(f) + assert result["total_messages"] == 1 + assert len(result["messages"]) == 1 + assert result["messages"][0]["subject"] == "Test delivery" + + +def test_deliver_to_inbox_file_missing_file(tmp_path, _noop_inbox_lock): + """Returns failure when inbox file does not exist.""" + inbox_file = tmp_path / "nonexistent.json" + + email_data = { + "from": "@sender", + "to": "@target", + "subject": "Test", + "message": "Body", + "timestamp": "2026-04-01 10:00:00", + } + + success, error, reply_id = deliver_to_inbox_file(inbox_file, email_data) + + assert success is False + assert "inbox not found" in error + assert reply_id == "" + + +def test_deliver_to_inbox_file_preserves_existing_messages(tmp_path, _noop_inbox_lock): + """New message is prepended to existing messages.""" + inbox_file = tmp_path / "inbox.json" + inbox_data = { + "mailbox": "inbox", + "total_messages": 1, + "unread_count": 0, + "messages": [{"id": "existing", "subject": "Old email", "status": "opened"}], + } + inbox_file.write_text(json.dumps(inbox_data, indent=2), encoding="utf-8") + + email_data = { + "from": "@sender", + "to": "@target", + "subject": "New email", + "message": "New body", + "timestamp": "2026-04-01 12:00:00", + "status": "new", + } + + success, error, reply_id = deliver_to_inbox_file(inbox_file, email_data) + + assert success is True + with open(inbox_file, "r", encoding="utf-8") as f: + result = json.load(f) + assert result["total_messages"] == 2 + assert result["messages"][0]["subject"] == "New email" + assert result["messages"][1]["subject"] == "Old email" + + +# ============================================================== +# push_dashboard_update tests +# ============================================================== + + +def test_push_dashboard_update_happy_path(tmp_path): + """Successful dashboard push returns True.""" + branch_path = tmp_path / "trigger" + inbox_dir = branch_path / ".ai_mail.local" + inbox_dir.mkdir(parents=True) + inbox_file = inbox_dir / "inbox.json" + inbox_data = { + "messages": [ + {"id": "m1", "status": "new", "timestamp": "2026-04-01 10:00:00"}, + {"id": "m2", "status": "opened", "timestamp": "2026-04-01 09:00:00"}, + ] + } + inbox_file.write_text(json.dumps(inbox_data), encoding="utf-8") + + mock_write = MagicMock(return_value=True) + + with patch.object(dashboard_mod, "_get_write_section", return_value=mock_write): + result = push_dashboard_update(branch_path) + + assert result is True + mock_write.assert_called_once() + section_data = mock_write.call_args[0][1] + assert section_data == "ai_mail" + + +def test_push_dashboard_update_no_inbox(tmp_path): + """Returns True with zero stats when no inbox exists.""" + branch_path = tmp_path / "empty_branch" + branch_path.mkdir() + + mock_write = MagicMock(return_value=True) + + with patch.object(dashboard_mod, "_get_write_section", return_value=mock_write): + result = push_dashboard_update(branch_path) + + assert result is True + mock_write.assert_called_once() + section_data = mock_write.call_args[0][2] + assert section_data["new"] == 0 + assert section_data["total"] == 0 + + +def test_push_dashboard_update_catches_exceptions(tmp_path): + """Returns False on any exception (never raises).""" + branch_path = tmp_path / "broken" + branch_path.mkdir() + + with patch.object(dashboard_mod, "_get_write_section", side_effect=RuntimeError("broken")): + result = push_dashboard_update(branch_path) + + assert result is False + + +# ============================================================== +# resolve_inbox_target tests +# ============================================================== + + +def test_resolve_inbox_target_explicit_branch(tmp_path): + """Resolves inbox for an explicit @branch target.""" + branch_info = {"path": str(tmp_path / "flow"), "name": "FLOW"} + mock_get_branch = MagicMock(return_value=branch_info) + mock_get_user = MagicMock() + + success, result = resolve_inbox_target("@flow", tmp_path, mock_get_branch, mock_get_user) + + assert success is True + assert result["target_branch"] == "@flow" + assert result["display_name"] == "FLOW" + assert result["error"] is None + assert result["inbox_file"] == Path(tmp_path / "flow" / ".ai_mail.local" / "inbox.json") + mock_get_branch.assert_called_once_with("@flow") + mock_get_user.assert_not_called() + + +def test_resolve_inbox_target_unknown_branch(): + """Returns failure for unknown branch.""" + mock_get_branch = MagicMock(return_value=None) + mock_get_user = MagicMock() + + success, result = resolve_inbox_target("@nonexistent", Path("/repo"), mock_get_branch, mock_get_user) + + assert success is False + assert "Unknown branch" in result["error"] + + +def test_resolve_inbox_target_no_args_uses_current_user(tmp_path): + """Uses current user detection when no explicit target is provided.""" + user_info = { + "mailbox_path": str(tmp_path / ".ai_mail.local"), + "display_name": "TRIGGER", + } + mock_get_branch = MagicMock() + mock_get_user = MagicMock(return_value=user_info) + + success, result = resolve_inbox_target(None, tmp_path, mock_get_branch, mock_get_user) + + assert success is True + assert result["target_branch"] is None + assert result["display_name"] == "TRIGGER" + assert result["inbox_file"] == Path(tmp_path / ".ai_mail.local" / "inbox.json") + mock_get_branch.assert_not_called() + mock_get_user.assert_called_once() + + +def test_resolve_inbox_target_non_at_arg_uses_current_user(tmp_path): + """Non-@ argument is treated as no target (uses current user).""" + user_info = { + "mailbox_path": str(tmp_path / ".ai_mail.local"), + "display_name": "TRIGGER", + } + mock_get_branch = MagicMock() + mock_get_user = MagicMock(return_value=user_info) + + success, result = resolve_inbox_target("some_arg", tmp_path, mock_get_branch, mock_get_user) + + assert success is True + assert result["target_branch"] is None + mock_get_user.assert_called_once() + + +def test_resolve_inbox_target_relative_path_resolved(tmp_path): + """Relative branch path is resolved against repo_root.""" + branch_info = {"path": "src/flow", "name": "FLOW"} + mock_get_branch = MagicMock(return_value=branch_info) + mock_get_user = MagicMock() + + success, result = resolve_inbox_target("@flow", tmp_path, mock_get_branch, mock_get_user) + + assert success is True + resolved_inbox = result["inbox_file"] + assert resolved_inbox.is_absolute() diff --git a/src/aipass/ai_mail/tests/test_purge.py b/src/aipass/ai_mail/tests/test_purge.py new file mode 100644 index 00000000..b2df3044 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_purge.py @@ -0,0 +1,202 @@ +"""Tests for sent/deleted auto-purge handler -- purge_sent_folder, purge_deleted_folder, run_purge.""" + +import json +import os +import pytest +from unittest.mock import patch + +import aipass.ai_mail.apps.handlers.email.purge as purge_mod +from aipass.ai_mail.apps.handlers.email.purge import ( + purge_sent_folder, + purge_deleted_folder, + run_purge, +) + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(): + """Prevent log_operation from writing real JSON files during tests.""" + with patch("aipass.ai_mail.apps.handlers.email.purge.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +# ---- Helper -------------------------------------------------- + + +def _populate_folder(folder_path, count): + """Create count JSON files in folder_path with staggered mtimes. + + Files are named email_000.json through email_{count-1}.json. + Each file gets a slightly different mtime so sorting by mtime is deterministic. + """ + folder_path.mkdir(parents=True, exist_ok=True) + for i in range(count): + email_file = folder_path / f"email_{i:03d}.json" + email_data = { + "id": f"msg-{i:03d}", + "from": "@sender", + "to": "@recipient", + "subject": f"Email {i}", + "message": f"Body {i}", + "timestamp": f"2026-01-01 12:{i:02d}:00", + } + email_file.write_text(json.dumps(email_data), encoding="utf-8") + # Stagger mtimes so sorting is deterministic (newer files have later mtime) + base_time = 1700000000.0 + i + os.utime(str(email_file), (base_time, base_time)) + + +# ---- purge_sent_folder tests --------------------------------- + + +def test_purge_sent_folder_no_folder(tmp_path): + """Returns success with 0 purged when sent folder does not exist.""" + result = purge_sent_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 0 + + +def test_purge_sent_folder_below_threshold(tmp_path): + """Returns success with 0 purged when file count is at or below threshold.""" + _populate_folder(tmp_path / "sent", 10) + + result = purge_sent_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 0 + assert "Below threshold" in result["message"] + + +def test_purge_sent_folder_above_threshold_vectorize_success(tmp_path, monkeypatch): + """Purges oldest files when count exceeds threshold and vectorization succeeds.""" + _populate_folder(tmp_path / "sent", 13) + + monkeypatch.setattr( + purge_mod, "_vectorize_emails", lambda emails, folder_type: {"success": True, "count": len(emails)} + ) + + result = purge_sent_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 3 # 13 - 10 = 3 files purged + assert result["vectorized"] is True + + # Verify 10 files remain + remaining = list((tmp_path / "sent").glob("*.json")) + assert len(remaining) == 10 + + +def test_purge_sent_folder_above_threshold_vectorize_fails(tmp_path, monkeypatch): + """Preserves all files when vectorization fails.""" + _populate_folder(tmp_path / "sent", 13) + + monkeypatch.setattr( + purge_mod, "_vectorize_emails", lambda emails, folder_type: {"success": False, "error": "timeout"} + ) + + result = purge_sent_folder(tmp_path) + + assert result["success"] is False + assert result["purged_count"] == 0 + assert result["vectorized"] is False + + # All 13 files should still exist + remaining = list((tmp_path / "sent").glob("*.json")) + assert len(remaining) == 13 + + +# ---- purge_deleted_folder tests ------------------------------ + + +def test_purge_deleted_folder_no_folder(tmp_path): + """Returns success with 0 purged when deleted folder does not exist.""" + result = purge_deleted_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 0 + + +def test_purge_deleted_folder_below_threshold(tmp_path): + """Returns success with 0 purged when file count is at or below threshold.""" + _populate_folder(tmp_path / "deleted", 5) + + result = purge_deleted_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 0 + + +def test_purge_deleted_folder_above_threshold(tmp_path, monkeypatch): + """Purges oldest files from deleted folder when count exceeds threshold.""" + _populate_folder(tmp_path / "deleted", 15) + + monkeypatch.setattr( + purge_mod, "_vectorize_emails", lambda emails, folder_type: {"success": True, "count": len(emails)} + ) + + result = purge_deleted_folder(tmp_path) + + assert result["success"] is True + assert result["purged_count"] == 5 # 15 - 10 = 5 files purged + + remaining = list((tmp_path / "deleted").glob("*.json")) + assert len(remaining) == 10 + + +# ---- run_purge tests ----------------------------------------- + + +def test_run_purge_both_below_threshold(tmp_path): + """Both folders below threshold returns success with 0 purged.""" + _populate_folder(tmp_path / "sent", 3) + _populate_folder(tmp_path / "deleted", 2) + + result = run_purge(tmp_path) + + assert result["success"] is True + assert result["sent"]["purged_count"] == 0 + assert result["deleted"]["purged_count"] == 0 + + +def test_run_purge_no_folders(tmp_path): + """No folders at all returns success.""" + result = run_purge(tmp_path) + + assert result["success"] is True + assert result["sent"]["purged_count"] == 0 + assert result["deleted"]["purged_count"] == 0 + + +def test_run_purge_mixed_results(tmp_path, monkeypatch): + """Sent over threshold and deleted below returns combined result.""" + _populate_folder(tmp_path / "sent", 12) + _populate_folder(tmp_path / "deleted", 5) + + monkeypatch.setattr( + purge_mod, "_vectorize_emails", lambda emails, folder_type: {"success": True, "count": len(emails)} + ) + + result = run_purge(tmp_path) + + assert result["success"] is True + assert result["sent"]["purged_count"] == 2 # 12 - 10 + assert result["deleted"]["purged_count"] == 0 + + +def test_run_purge_failure_propagates(tmp_path, monkeypatch): + """Overall success is False when either folder purge fails.""" + _populate_folder(tmp_path / "sent", 15) + + monkeypatch.setattr( + purge_mod, "_vectorize_emails", lambda emails, folder_type: {"success": False, "error": "broken"} + ) + + result = run_purge(tmp_path) + + assert result["success"] is False + assert result["sent"]["success"] is False diff --git a/src/aipass/ai_mail/tests/test_reply.py b/src/aipass/ai_mail/tests/test_reply.py new file mode 100644 index 00000000..f289c87c --- /dev/null +++ b/src/aipass/ai_mail/tests/test_reply.py @@ -0,0 +1,270 @@ +"""Tests for email reply handler -- get_email_by_id and send_reply.""" + +import json +import pytest +from unittest.mock import patch + + +from aipass.ai_mail.apps.handlers.email.reply import ( + get_email_by_id, + send_reply, +) + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(): + """Prevent log_operation from writing real JSON files during tests.""" + with patch("aipass.ai_mail.apps.handlers.email.reply.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +# ---- get_email_by_id tests ----------------------------------- + + +def test_get_email_by_id_found(tmp_path): + """Returns matching message dict when ID exists in inbox.""" + inbox_file = tmp_path / "inbox.json" + inbox_data = { + "messages": [ + {"id": "abc123", "subject": "First", "status": "new"}, + {"id": "def456", "subject": "Second", "status": "opened"}, + ] + } + inbox_file.write_text(json.dumps(inbox_data), encoding="utf-8") + + result = get_email_by_id(inbox_file, "def456") + + assert result is not None + assert result["id"] == "def456" + assert result["subject"] == "Second" + assert result["status"] == "opened" + + +def test_get_email_by_id_not_found(tmp_path): + """Returns None when no message matches the ID.""" + inbox_file = tmp_path / "inbox.json" + inbox_data = { + "messages": [ + {"id": "abc123", "subject": "Only one"}, + ] + } + inbox_file.write_text(json.dumps(inbox_data), encoding="utf-8") + + result = get_email_by_id(inbox_file, "nonexistent") + + assert result is None + + +def test_get_email_by_id_missing_file(tmp_path): + """Returns None when inbox file does not exist.""" + inbox_file = tmp_path / "does_not_exist.json" + + result = get_email_by_id(inbox_file, "abc123") + + assert result is None + + +def test_get_email_by_id_corrupt_json(tmp_path): + """Returns None when inbox file contains invalid JSON.""" + inbox_file = tmp_path / "inbox.json" + inbox_file.write_text("{broken json!!!", encoding="utf-8") + + result = get_email_by_id(inbox_file, "abc123") + + assert result is None + + +def test_get_email_by_id_empty_messages(tmp_path): + """Returns None when messages list is empty.""" + inbox_file = tmp_path / "inbox.json" + inbox_data = {"messages": []} + inbox_file.write_text(json.dumps(inbox_data), encoding="utf-8") + + result = get_email_by_id(inbox_file, "abc123") + + assert result is None + + +def test_get_email_by_id_no_messages_key(tmp_path): + """Returns None when inbox JSON has no messages key.""" + inbox_file = tmp_path / "inbox.json" + inbox_data = {"mailbox": "inbox"} + inbox_file.write_text(json.dumps(inbox_data), encoding="utf-8") + + result = get_email_by_id(inbox_file, "abc123") + + assert result is None + + +# ---- send_reply tests ---------------------------------------- + + +def _make_original_email( + *, + msg_id: str = "orig-001", + sender: str = "@devpulse", + subject: str = "Original subject", + reply_to: str | None = None, + dispatched_to: str | None = None, +) -> dict: + """Build a minimal original email dict for send_reply tests.""" + email = { + "id": msg_id, + "from": sender, + "subject": subject, + "status": "opened", + } + if reply_to is not None: + email["reply_to"] = reply_to + if dispatched_to is not None: + email["dispatched_to"] = dispatched_to + return email + + +# Patch paths for lazy imports inside send_reply function body +_PATCH_BRANCH_DETECTION = "aipass.ai_mail.apps.handlers.users.branch_detection.get_branch_info_from_registry" +_PATCH_DELIVERY = "aipass.ai_mail.apps.handlers.email.delivery.deliver_email_to_branch" +_PATCH_ALL_BRANCHES = "aipass.ai_mail.apps.handlers.registry.read.get_all_branches" +_PATCH_CLOSE_ARCHIVE = "aipass.ai_mail.apps.handlers.email.inbox_cleanup.mark_as_closed_and_archive" + + +def test_send_reply_happy_path(tmp_path): + """Successful reply returns (True, message, reply_id).""" + from_branch_path = tmp_path / "trigger" + from_branch_path.mkdir() + + sender_info = {"email": "@trigger", "name": "TRIGGER"} + target_branch = {"email": "@devpulse", "name": "DEVPULSE", "path": str(tmp_path / "devpulse")} + original = _make_original_email() + + with ( + patch(_PATCH_BRANCH_DETECTION, return_value=sender_info), + patch(_PATCH_DELIVERY, return_value=(True, "")), + patch(_PATCH_ALL_BRANCHES, return_value=[target_branch]), + patch(_PATCH_CLOSE_ARCHIVE, return_value=(True, "closed")), + ): + success, message, reply_id = send_reply(from_branch_path, original, "Thanks!") + + assert success is True + assert reply_id is not None + assert "Reply sent" in message + + # Verify sent file was created + sent_folder = from_branch_path / ".ai_mail.local" / "sent" + assert sent_folder.exists() + sent_files = list(sent_folder.glob("*.json")) + assert len(sent_files) == 1 + + with open(sent_files[0], "r", encoding="utf-8") as f: + sent_data = json.load(f) + assert sent_data["from"] == "@trigger" + assert sent_data["to"] == "@devpulse" + assert sent_data["subject"].startswith("RE:") + assert sent_data["message"] == "Thanks!" + assert sent_data["in_reply_to"] == "orig-001" + + +def test_send_reply_no_sender_info(tmp_path): + """Returns failure when sender branch cannot be detected.""" + from_branch_path = tmp_path / "unknown" + from_branch_path.mkdir() + original = _make_original_email() + + with patch(_PATCH_BRANCH_DETECTION, return_value=None): + success, message, reply_id = send_reply(from_branch_path, original, "Reply text") + + assert success is False + assert "Could not detect" in message + assert reply_id is None + + +def test_send_reply_unknown_recipient(tmp_path): + """Returns failure when recipient branch is not found in registry.""" + from_branch_path = tmp_path / "trigger" + from_branch_path.mkdir() + sender_info = {"email": "@trigger", "name": "TRIGGER"} + original = _make_original_email(sender="@unknown_branch") + + with ( + patch(_PATCH_BRANCH_DETECTION, return_value=sender_info), + patch(_PATCH_ALL_BRANCHES, return_value=[]), + ): + success, message, reply_id = send_reply(from_branch_path, original, "Reply text") + + assert success is False + assert "Could not find branch" in message + assert reply_id is None + + +def test_send_reply_identity_mismatch_raises(tmp_path): + """Raises RuntimeError when dispatched_to does not match current sender.""" + from_branch_path = tmp_path / "wrong_branch" + from_branch_path.mkdir() + sender_info = {"email": "@wrong_branch", "name": "WRONG"} + original = _make_original_email(dispatched_to="@correct_branch") + + with ( + patch(_PATCH_BRANCH_DETECTION, return_value=sender_info), + pytest.raises(RuntimeError, match="IDENTITY MISMATCH"), + ): + send_reply(from_branch_path, original, "Reply text") + + +def test_send_reply_uses_reply_to_field(tmp_path): + """Reply goes to reply_to address when present, not the from address.""" + from_branch_path = tmp_path / "trigger" + from_branch_path.mkdir() + sender_info = {"email": "@trigger", "name": "TRIGGER"} + target_branch = {"email": "@flow", "name": "FLOW", "path": str(tmp_path / "flow")} + original = _make_original_email(sender="@devpulse", reply_to="@flow") + + deliver_calls = [] + + def mock_deliver(to_branch, email_data): + """Capture delivery arguments.""" + deliver_calls.append((to_branch, email_data)) + return (True, "") + + with ( + patch(_PATCH_BRANCH_DETECTION, return_value=sender_info), + patch(_PATCH_DELIVERY, side_effect=mock_deliver), + patch(_PATCH_ALL_BRANCHES, return_value=[target_branch]), + patch(_PATCH_CLOSE_ARCHIVE, return_value=(True, "closed")), + ): + success, message, reply_id = send_reply(from_branch_path, original, "Thanks!") + + assert success is True + assert len(deliver_calls) == 1 + assert deliver_calls[0][0] == "@flow" + + +def test_send_reply_re_prefix_not_duplicated(tmp_path): + """Subject already starting with RE: does not get double-prefixed.""" + from_branch_path = tmp_path / "trigger" + from_branch_path.mkdir() + sender_info = {"email": "@trigger", "name": "TRIGGER"} + target_branch = {"email": "@devpulse", "name": "DEVPULSE", "path": str(tmp_path / "devpulse")} + original = _make_original_email(subject="RE: Already replied") + + deliver_calls = [] + + def mock_deliver(to_branch, email_data): + """Capture delivery arguments.""" + deliver_calls.append((to_branch, email_data)) + return (True, "") + + with ( + patch(_PATCH_BRANCH_DETECTION, return_value=sender_info), + patch(_PATCH_DELIVERY, side_effect=mock_deliver), + patch(_PATCH_ALL_BRANCHES, return_value=[target_branch]), + patch(_PATCH_CLOSE_ARCHIVE, return_value=(True, "closed")), + ): + success, _message, _reply_id = send_reply(from_branch_path, original, "Thanks!") + + assert success is True + # Should keep "RE: Already replied", not "RE: RE: Already replied" + assert deliver_calls[0][1]["subject"] == "RE: Already replied" diff --git a/src/aipass/ai_mail/tests/test_send_helpers.py b/src/aipass/ai_mail/tests/test_send_helpers.py new file mode 100644 index 00000000..5b16014b --- /dev/null +++ b/src/aipass/ai_mail/tests/test_send_helpers.py @@ -0,0 +1,359 @@ +"""Tests for email send handler -- send_to_single, send_to_broadcast, collect_interactive_input, +and resolve_dispatch_target from send_args.""" + +import pytest +from unittest.mock import patch, MagicMock + +from aipass.ai_mail.apps.handlers.email.send import ( + send_to_single, + send_to_broadcast, + collect_interactive_input, +) +from aipass.ai_mail.apps.handlers.email.send_args import ( + resolve_dispatch_target, +) + + +# ---- Fixtures ------------------------------------------------ + + +@pytest.fixture(autouse=True) +def _silence_json_handler(): + """Prevent log_operation from writing real JSON files during tests.""" + with patch("aipass.ai_mail.apps.handlers.email.send.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +@pytest.fixture(autouse=True) +def _silence_send_args_json_handler(): + """Prevent log_operation in send_args from writing real JSON files.""" + with patch("aipass.ai_mail.apps.handlers.email.send_args.json_handler") as mock_jh: + mock_jh.log_operation.return_value = True + yield mock_jh + + +# ---- send_to_single tests ------------------------------------ + + +def _make_user_info() -> dict: + """Build a minimal user_info dict for send tests.""" + return { + "email_address": "@trigger", + "display_name": "TRIGGER", + "mailbox_path": "/tmp/trigger/.ai_mail.local", + "timestamp_format": "%Y-%m-%d %H:%M:%S", + } + + +def test_send_to_single_happy_path(): + """Successful single send returns (True, None).""" + mock_create = MagicMock(return_value="/tmp/email_file.json") + mock_load = MagicMock(return_value={"subject": "Test", "message": "Body"}) + mock_deliver = MagicMock(return_value=(True, "")) + mock_callback = MagicMock() + mock_log = MagicMock() + mock_update = MagicMock() + + success, error = send_to_single( + to_branch="@backup", + subject="Test subject", + message="Test body", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=mock_callback, + log_operation_fn=mock_log, + update_central_fn=mock_update, + ) + + assert success is True + assert error is None + mock_create.assert_called_once() + mock_load.assert_called_once_with("/tmp/email_file.json") + mock_deliver.assert_called_once() + mock_log.assert_called_once_with("email_sent", {"to": "@backup", "subject": "Test subject", "auto_execute": False}) + + +def test_send_to_single_load_fails(): + """Returns (False, error) when email file cannot be loaded.""" + mock_create = MagicMock(return_value="/tmp/email_file.json") + mock_load = MagicMock(return_value=None) + mock_deliver = MagicMock() + mock_log = MagicMock() + + success, error = send_to_single( + to_branch="@backup", + subject="Test", + message="Body", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=None, + log_operation_fn=mock_log, + update_central_fn=None, + ) + + assert success is False + assert error is not None + assert "could not be loaded" in error + mock_deliver.assert_not_called() + + +def test_send_to_single_delivery_fails(): + """Returns (False, error) when delivery function reports failure.""" + mock_create = MagicMock(return_value="/tmp/email_file.json") + mock_load = MagicMock(return_value={"subject": "Test", "message": "Body"}) + mock_deliver = MagicMock(return_value=(False, "Branch offline")) + mock_log = MagicMock() + + success, error = send_to_single( + to_branch="@backup", + subject="Test", + message="Body", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=None, + log_operation_fn=mock_log, + update_central_fn=None, + ) + + assert success is False + assert error == "Branch offline" + + +def test_send_to_single_sets_auto_execute(): + """auto_execute flag is set on email_data before delivery.""" + captured_data = {} + + def mock_deliver(to, data, on_delivered=None): + """Capture delivery data for assertion.""" + captured_data.update(data) + return (True, "") + + mock_create = MagicMock(return_value="/tmp/email.json") + mock_load = MagicMock(return_value={"subject": "Test", "message": "Body"}) + + send_to_single( + to_branch="@flow", + subject="Test", + message="Body", + user_info=_make_user_info(), + auto_execute=True, + no_memory_save=True, + reply_to=None, + dispatched_to="@flow", + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=None, + log_operation_fn=MagicMock(), + update_central_fn=None, + ) + + assert captured_data["auto_execute"] is True + assert captured_data["dispatched_to"] == "@flow" + assert captured_data["no_memory_save"] is True + + +# ---- send_to_broadcast tests --------------------------------- + + +def test_send_to_broadcast_happy_path(): + """Successful broadcast returns (True, success_count, total, results).""" + branches = [ + {"email": "@flow", "name": "FLOW"}, + {"email": "@backup", "name": "BACKUP"}, + ] + mock_create = MagicMock(return_value="/tmp/broadcast.json") + mock_load = MagicMock(return_value={"subject": "Announce", "message": "Hello all"}) + mock_deliver = MagicMock(return_value=(True, "")) + mock_log = MagicMock() + + ok, success_count, total, results = send_to_broadcast( + subject="Announce", + message="Hello all", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + branches=branches, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=None, + log_operation_fn=mock_log, + update_central_fn=None, + ) + + assert ok is True + assert success_count == 2 + assert total == 2 + assert isinstance(results, list) + assert len(results) == 2 + + +def test_send_to_broadcast_load_fails(): + """Returns failure when email file cannot be loaded.""" + branches = [{"email": "@flow", "name": "FLOW"}] + mock_create = MagicMock(return_value="/tmp/broadcast.json") + mock_load = MagicMock(return_value=None) + mock_log = MagicMock() + + ok, success_count, total, error = send_to_broadcast( + subject="Announce", + message="Hello", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + branches=branches, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=MagicMock(), + on_delivered_callback=None, + log_operation_fn=mock_log, + update_central_fn=None, + ) + + assert ok is False + assert success_count == 0 + assert "could not be loaded" in error + + +def test_send_to_broadcast_partial_failure(): + """Partial delivery failure returns correct counts.""" + branches = [ + {"email": "@flow", "name": "FLOW"}, + {"email": "@backup", "name": "BACKUP"}, + {"email": "@memory", "name": "MEMORY"}, + ] + mock_create = MagicMock(return_value="/tmp/broadcast.json") + mock_load = MagicMock(return_value={"subject": "Test", "message": "Body"}) + # First and third succeed, second fails + mock_deliver = MagicMock(side_effect=[(True, ""), (False, "offline"), (True, "")]) + mock_log = MagicMock() + + ok, success_count, total, results = send_to_broadcast( + subject="Test", + message="Body", + user_info=_make_user_info(), + auto_execute=False, + no_memory_save=False, + reply_to=None, + dispatched_to=None, + branches=branches, + create_email_file_fn=mock_create, + load_email_file_fn=mock_load, + deliver_email_to_branch_fn=mock_deliver, + on_delivered_callback=None, + log_operation_fn=mock_log, + update_central_fn=None, + ) + + assert ok is True # At least one succeeded + assert success_count == 2 + assert total == 3 + assert results[1][1] is False # Second branch failed + assert results[1][2] == "offline" + + +# ---- collect_interactive_input tests -------------------------- + + +def test_collect_interactive_input_cancelled_on_eof(): + """Returns None when input raises EOFError (cancelled).""" + branches = [{"email": "@flow", "name": "FLOW"}] + + with patch("builtins.input", side_effect=EOFError): + result = collect_interactive_input(branches) + + assert result is None + + +def test_collect_interactive_input_cancelled_on_keyboard_interrupt(): + """Returns None when input raises KeyboardInterrupt.""" + branches = [{"email": "@flow", "name": "FLOW"}] + + with patch("builtins.input", side_effect=KeyboardInterrupt): + result = collect_interactive_input(branches) + + assert result is None + + +def test_collect_interactive_input_invalid_selection(): + """Returns None when user enters non-numeric selection.""" + branches = [{"email": "@flow", "name": "FLOW"}] + + with patch("builtins.input", return_value="abc"): + result = collect_interactive_input(branches) + + assert result is None + + +# ---- resolve_dispatch_target tests ---------------------------- + + +def test_resolve_dispatch_target_no_auto_execute(): + """Returns None when auto_execute is False.""" + result = resolve_dispatch_target("@flow", False) + + assert result is None + + +def test_resolve_dispatch_target_email_address(): + """Returns the branch email when auto_execute is True and branch starts with @.""" + result = resolve_dispatch_target("@flow", True) + + assert result == "@flow" + + +def test_resolve_dispatch_target_path_with_registry_lookup(): + """Returns registry email when path resolves via get_branch_info_fn.""" + mock_fn = MagicMock(return_value={"email": "@trigger", "name": "TRIGGER"}) + + result = resolve_dispatch_target("/home/user/trigger", True, get_branch_info_fn=mock_fn) + + assert result == "@trigger" + mock_fn.assert_called_once() + + +def test_resolve_dispatch_target_path_without_registry(): + """Returns fallback @dirname when no registry function provided.""" + result = resolve_dispatch_target("/home/user/flow", True, get_branch_info_fn=None) + + assert result == "@flow" + + +def test_resolve_dispatch_target_path_registry_not_found(): + """Returns fallback @dirname when registry lookup returns None.""" + mock_fn = MagicMock(return_value=None) + + result = resolve_dispatch_target("/home/user/backup", True, get_branch_info_fn=mock_fn) + + assert result == "@backup" + + +def test_resolve_dispatch_target_tilde_path(): + """Handles ~ prefixed paths by extracting the directory name.""" + result = resolve_dispatch_target("~/Projects/flow", True, get_branch_info_fn=None) + + assert result == "@flow"