from __future__ import annotations from unittest.mock import AsyncMock, MagicMock, patch from datetime import datetime, timezone import pytest from howl.db.queries import SenderMatch from howl.llm.schemas import EmailClassification from howl.pipeline import classifier from howl.pipeline.actions import ActionExecutor from howl.pipeline.processor import EmailProcessor # --------------------------------------------------------------------------- # ActionExecutor tests (dry_run mode — no actual HTTP calls) # --------------------------------------------------------------------------- def test_action_executor_dry_run_does_not_call_graph(settings, mock_graph): from howl.db.models import EmailLog assert settings.dry_run is True # conftest sets dry_run=True executor = ActionExecutor(mock_graph, settings) log_entry = MagicMock(spec=EmailLog) log_entry.mailbox = "test@example.com" log_entry.graph_message_id = "msg-001" log_entry.sender_name = "Alice" log_entry.sender_address = "alice@test.com" log_entry.subject = "Test" executor.execute(log_entry, "move_customer") mock_graph.move_message.assert_not_called() mock_graph.get_or_create_folder.assert_not_called() def test_action_executor_move_customer_live(settings, mock_graph): settings = settings.model_copy(update={"dry_run": False}) from howl.db.models import EmailLog executor = ActionExecutor(mock_graph, settings) log_entry = MagicMock(spec=EmailLog) log_entry.mailbox = "test@example.com" log_entry.graph_message_id = "msg-001" log_entry.sender_name = "Alice" log_entry.sender_address = "alice@test.com" log_entry.subject = "Test" executor.execute(log_entry, "move_customer") mock_graph.get_or_create_folder.assert_called_once_with("test@example.com", settings.folder_customers) mock_graph.move_message.assert_called_once() def test_action_executor_escalate_sends_notification(settings, mock_graph): settings = settings.model_copy(update={ "dry_run": False, "notification_webhook_url": "https://hooks.example.com/notify", }) from howl.db.models import EmailLog executor = ActionExecutor(mock_graph, settings) log_entry = MagicMock(spec=EmailLog) log_entry.mailbox = "test@example.com" log_entry.graph_message_id = "msg-escalate" log_entry.sender_name = "Bob" log_entry.sender_address = "bob@client.com" log_entry.subject = "URGENT" with patch("httpx.post") as mock_post: mock_post.return_value = MagicMock(status_code=200, raise_for_status=MagicMock()) executor.execute(log_entry, "escalate") mock_graph.flag_message.assert_called_once() mock_post.assert_called_once() call_kwargs = mock_post.call_args assert call_kwargs[0][0] == "https://hooks.example.com/notify" assert "text" in call_kwargs[1]["json"] assert call_kwargs[1]["timeout"] == 10 # --------------------------------------------------------------------------- # Full pipeline integration (all external calls mocked) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_processor_process_message_happy_path( settings, mock_graph, mock_llm, customer_message, customer_match ): """Full pipeline: message → DB lookup → LLM → classify → action → log.""" processor = EmailProcessor(mock_graph, mock_llm, settings) inserted_log = MagicMock() inserted_log.id = "log-uuid-001" inserted_log.retry_count = 0 with ( patch("howl.pipeline.processor.db_engine") as mock_db_engine, patch("howl.pipeline.processor.queries") as mock_queries, ): # Set up async session context manager mock_session = AsyncMock() mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False) mock_queries.is_already_processed = AsyncMock(return_value=False) mock_queries.insert_email_log = AsyncMock(return_value=inserted_log) mock_queries.lookup_sender = AsyncMock(return_value=customer_match) mock_queries.update_email_log = AsyncMock() await processor.process_message(customer_message) # Verify DB was consulted for idempotency mock_queries.is_already_processed.assert_called_once_with(mock_session, customer_message.id) # Verify email log was created mock_queries.insert_email_log.assert_called_once() # Verify sender was looked up mock_queries.lookup_sender.assert_called_once_with(mock_session, customer_message.sender_address) # Verify LLM was called mock_llm.classify.assert_called_once() # Verify log was updated with completed status final_update_calls = [ c for c in mock_queries.update_email_log.call_args_list if c.kwargs.get("status") == "completed" or (len(c.args) > 2 and "completed" in str(c)) ] # At minimum update_email_log should have been called assert mock_queries.update_email_log.call_count >= 2 @pytest.mark.asyncio async def test_processor_skips_already_processed_message( settings, mock_graph, mock_llm, customer_message ): processor = EmailProcessor(mock_graph, mock_llm, settings) with ( patch("howl.pipeline.processor.db_engine") as mock_db_engine, patch("howl.pipeline.processor.queries") as mock_queries, ): mock_session = AsyncMock() mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False) mock_queries.is_already_processed = AsyncMock(return_value=True) await processor.process_message(customer_message) # Should not proceed past idempotency check mock_llm.classify.assert_not_called() mock_graph.move_message.assert_not_called() @pytest.mark.asyncio async def test_processor_handles_llm_failure_gracefully( settings, mock_graph, mock_llm, customer_message, customer_match ): """If LLM throws, the pipeline should catch it and log failure — not crash.""" mock_llm.classify.side_effect = RuntimeError("Anthropic API timeout") processor = EmailProcessor(mock_graph, mock_llm, settings) inserted_log = MagicMock() inserted_log.id = "log-uuid-002" inserted_log.retry_count = 0 with ( patch("howl.pipeline.processor.db_engine") as mock_db_engine, patch("howl.pipeline.processor.queries") as mock_queries, ): mock_session = AsyncMock() mock_db_engine.async_session.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_db_engine.async_session.return_value.__aexit__ = AsyncMock(return_value=False) mock_queries.is_already_processed = AsyncMock(return_value=False) mock_queries.insert_email_log = AsyncMock(return_value=inserted_log) mock_queries.lookup_sender = AsyncMock(return_value=customer_match) mock_queries.update_email_log = AsyncMock() # Should NOT raise await processor.process_message(customer_message) # Should have logged failure failed_updates = [ c for c in mock_queries.update_email_log.call_args_list if c.kwargs.get("status") == "failed" ] assert len(failed_updates) >= 1