feat(ai_mail): test: cover all 44 untested functions per TDPLAN-0003
Co-Authored-By: @ai_mail <ai_mail@aipass>
This commit is contained in:
AIOSAI
@@ -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()
|
||||
@@ -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
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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()
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user