from datetime import UTC, datetime, timedelta from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from app.models.whatsapp import TemplateStatus from app.services.whatsapp_sender import ( SessionWindowExpired, WhatsAppSender, WhatsAppSendError, ) PHONE_NUMBER_ID = "12345" RECIPIENT = "+31612345678" def _mock_account(): account = MagicMock() account.id = "wa-account-id" account.restaurant_id = "restaurant-id" account.phone_number_id = PHONE_NUMBER_ID account.is_active = True account.access_token_encrypted = b"encrypted-token-bytes" return account def _scalar_result(value): result = MagicMock() result.scalar_one_or_none.return_value = value return result def _mock_template(status: str = TemplateStatus.APPROVED.value): template = MagicMock() template.status = status return template def _response(status_code: int, payload: dict): request = httpx.Request( "POST", f"https://graph.facebook.com/v21.0/{PHONE_NUMBER_ID}/messages", ) return httpx.Response(status_code=status_code, json=payload, request=request) class TestWhatsAppSender: @pytest.mark.asyncio(loop_scope="session") async def test_send_text_success(self, mock_session): account = _mock_account() mock_session.execute.return_value = _scalar_result(account) client = AsyncMock() client.post.return_value = _response(200, {"messages": [{"id": "wamid.123"}]}) sender = WhatsAppSender(mock_session, client=client) valid_at = datetime.now(UTC) - timedelta(minutes=1) with patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ) as mock_decrypt: result = await sender.send_text( PHONE_NUMBER_ID, RECIPIENT, " Hello from WhatsApp ", last_customer_message_at=valid_at, ) mock_decrypt.assert_called_once_with(account.access_token_encrypted) client.post.assert_awaited_once() assert result["wamid"] == "wamid.123" _, kwargs = client.post.await_args assert kwargs["headers"]["Authorization"] == "Bearer test-access-token" assert kwargs["json"] == { "messaging_product": "whatsapp", "recipient_type": "individual", "to": RECIPIENT, "type": "text", "text": {"body": "Hello from WhatsApp"}, } @pytest.mark.asyncio(loop_scope="session") async def test_send_text_session_window_expired(self, mock_session): client = AsyncMock() sender = WhatsAppSender(mock_session, client=client) expired_at = datetime.now(UTC) - timedelta(hours=24, minutes=1) with pytest.raises(SessionWindowExpired): await sender.send_text( PHONE_NUMBER_ID, RECIPIENT, "Outside session window", last_customer_message_at=expired_at, ) mock_session.execute.assert_not_called() client.post.assert_not_awaited() @pytest.mark.asyncio(loop_scope="session") async def test_send_text_session_window_valid(self, mock_session): account = _mock_account() mock_session.execute.return_value = _scalar_result(account) client = AsyncMock() client.post.return_value = _response(200, {"messages": [{"id": "wamid.234"}]}) sender = WhatsAppSender(mock_session, client=client) valid_at = datetime.now(UTC) - timedelta(hours=23, minutes=59) with patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ): result = await sender.send_text( PHONE_NUMBER_ID, RECIPIENT, "Inside session window", last_customer_message_at=valid_at, ) assert result["wamid"] == "wamid.234" client.post.assert_awaited_once() @pytest.mark.asyncio(loop_scope="session") async def test_send_template_approved(self, mock_session): account = _mock_account() template = _mock_template() mock_session.execute.side_effect = [_scalar_result(account), _scalar_result(template)] client = AsyncMock() client.post.return_value = _response(200, {"messages": [{"id": "wamid.template"}]}) sender = WhatsAppSender(mock_session, client=client) components = [{"type": "body", "parameters": [{"type": "text", "text": "Ada"}]}] with patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ): result = await sender.send_template( PHONE_NUMBER_ID, RECIPIENT, "booking_confirmation", "en", components=components, ) assert result["wamid"] == "wamid.template" assert mock_session.execute.await_count == 2 _, kwargs = client.post.await_args assert kwargs["json"] == { "messaging_product": "whatsapp", "recipient_type": "individual", "to": RECIPIENT, "type": "template", "template": { "name": "booking_confirmation", "language": {"code": "en"}, "components": components, }, } @pytest.mark.asyncio(loop_scope="session") async def test_send_template_not_approved(self, mock_session): account = _mock_account() template = _mock_template(status=TemplateStatus.PENDING.value) mock_session.execute.side_effect = [_scalar_result(account), _scalar_result(template)] client = AsyncMock() sender = WhatsAppSender(mock_session, client=client) with ( patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ), pytest.raises(ValueError, match="not approved"), ): await sender.send_template( PHONE_NUMBER_ID, RECIPIENT, "booking_confirmation", "en", ) client.post.assert_not_awaited() @pytest.mark.asyncio(loop_scope="session") async def test_retry_on_rate_limit(self, mock_session): account = _mock_account() mock_session.execute.return_value = _scalar_result(account) client = AsyncMock() client.post.side_effect = [ _response(429, {"error": {"message": "rate limited"}}), _response(429, {"error": {"message": "rate limited"}}), _response(200, {"messages": [{"id": "wamid.retry"}]}), ] sender = WhatsAppSender(mock_session, client=client) valid_at = datetime.now(UTC) - timedelta(minutes=1) with ( patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ), patch( "app.services.whatsapp_sender.asyncio.sleep", new_callable=AsyncMock, ) as mock_sleep, ): result = await sender.send_text( PHONE_NUMBER_ID, RECIPIENT, "Retry me", last_customer_message_at=valid_at, ) assert result["wamid"] == "wamid.retry" assert client.post.await_count == 3 assert mock_sleep.await_count == 2 assert [call.args[0] for call in mock_sleep.await_args_list] == [1.0, 2.0] @pytest.mark.asyncio(loop_scope="session") async def test_permanent_error_no_retry(self, mock_session): account = _mock_account() mock_session.execute.return_value = _scalar_result(account) client = AsyncMock() client.post.return_value = _response( 400, {"error": {"message": "invalid recipient"}}, ) valid_at = datetime.now(UTC) - timedelta(minutes=1) sender = WhatsAppSender(mock_session, client=client) with ( patch( "app.services.whatsapp_sender.decrypt_token", return_value="test-access-token", ), patch( "app.services.whatsapp_sender.asyncio.sleep", new_callable=AsyncMock, ) as mock_sleep, pytest.raises( WhatsAppSendError, match="invalid recipient", ) as exc_info, ): await sender.send_text( PHONE_NUMBER_ID, RECIPIENT, "Will fail", last_customer_message_at=valid_at, ) assert exc_info.value.is_transient is False assert client.post.await_count == 1 mock_sleep.assert_not_awaited() def test_session_window_missing_timestamp(self, mock_session): sender = WhatsAppSender(mock_session, client=AsyncMock()) with pytest.raises(SessionWindowExpired): sender._ensure_session_window(None)