|
| 1 | +""" |
| 2 | +Unit tests for WhatsApp Web adapter (_whatsapp_web_adapter.py). |
| 3 | +
|
| 4 | +Tests the neonize adapter isolation layer with mocked neonize client. |
| 5 | +All tests are deterministic — no external WhatsApp calls. |
| 6 | +""" |
| 7 | + |
| 8 | +import os |
| 9 | +import sys |
| 10 | +import tempfile |
| 11 | +from unittest.mock import AsyncMock, MagicMock, patch |
| 12 | + |
| 13 | +import pytest |
| 14 | + |
| 15 | +# Ensure the wrapper package is importable |
| 16 | +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "praisonai")) |
| 17 | + |
| 18 | + |
| 19 | +# ── Adapter import ──────────────────────────────────────────────── |
| 20 | + |
| 21 | +class TestAdapterImport: |
| 22 | + """Test that the adapter module can be imported without neonize.""" |
| 23 | + |
| 24 | + def test_import_adapter_module(self): |
| 25 | + """Adapter module imports without neonize installed.""" |
| 26 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 27 | + assert WhatsAppWebAdapter is not None |
| 28 | + |
| 29 | + def test_default_creds_dir(self): |
| 30 | + """Default creds dir is ~/.praisonai/whatsapp.""" |
| 31 | + from praisonai.bots._whatsapp_web_adapter import DEFAULT_CREDS_DIR |
| 32 | + assert ".praisonai" in DEFAULT_CREDS_DIR |
| 33 | + assert "whatsapp" in DEFAULT_CREDS_DIR |
| 34 | + |
| 35 | + |
| 36 | +# ── Adapter construction ────────────────────────────────────────── |
| 37 | + |
| 38 | +class TestAdapterConstruction: |
| 39 | + """Test WhatsAppWebAdapter initialization.""" |
| 40 | + |
| 41 | + def test_default_construction(self): |
| 42 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 43 | + adapter = WhatsAppWebAdapter() |
| 44 | + assert adapter.is_connected is False |
| 45 | + assert adapter.self_jid is None |
| 46 | + assert ".praisonai" in adapter.creds_dir |
| 47 | + |
| 48 | + def test_custom_creds_dir(self): |
| 49 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 50 | + adapter = WhatsAppWebAdapter(creds_dir="/tmp/test-wa-creds") |
| 51 | + assert adapter.creds_dir == "/tmp/test-wa-creds" |
| 52 | + |
| 53 | + def test_custom_bot_name(self): |
| 54 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 55 | + adapter = WhatsAppWebAdapter(bot_name="my_bot") |
| 56 | + assert adapter._bot_name == "my_bot" |
| 57 | + |
| 58 | + def test_env_var_creds_dir(self): |
| 59 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 60 | + with patch.dict(os.environ, {"WHATSAPP_CREDS_DIR": "/tmp/env-creds"}): |
| 61 | + adapter = WhatsAppWebAdapter() |
| 62 | + assert adapter.creds_dir == "/tmp/env-creds" |
| 63 | + |
| 64 | + def test_callbacks_stored(self): |
| 65 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 66 | + on_msg = MagicMock() |
| 67 | + on_qr = MagicMock() |
| 68 | + on_conn = MagicMock() |
| 69 | + on_disc = MagicMock() |
| 70 | + adapter = WhatsAppWebAdapter( |
| 71 | + on_message=on_msg, |
| 72 | + on_qr=on_qr, |
| 73 | + on_connected=on_conn, |
| 74 | + on_disconnected=on_disc, |
| 75 | + ) |
| 76 | + assert adapter._on_message is on_msg |
| 77 | + assert adapter._on_qr is on_qr |
| 78 | + assert adapter._on_connected is on_conn |
| 79 | + assert adapter._on_disconnected is on_disc |
| 80 | + |
| 81 | + |
| 82 | +# ── Credential management ───────────────────────────────────────── |
| 83 | + |
| 84 | +class TestCredentialManagement: |
| 85 | + """Test credential path and session detection.""" |
| 86 | + |
| 87 | + def test_ensure_creds_dir_creates_directory(self): |
| 88 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 89 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 90 | + creds_path = os.path.join(tmpdir, "subdir", "wa") |
| 91 | + adapter = WhatsAppWebAdapter(creds_dir=creds_path) |
| 92 | + adapter._ensure_creds_dir() |
| 93 | + assert os.path.isdir(creds_path) |
| 94 | + |
| 95 | + def test_get_db_path(self): |
| 96 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 97 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 98 | + adapter = WhatsAppWebAdapter(creds_dir=tmpdir, bot_name="testbot") |
| 99 | + db_path = adapter._get_db_path() |
| 100 | + assert db_path.endswith("testbot.db") |
| 101 | + assert tmpdir in db_path |
| 102 | + |
| 103 | + def test_has_saved_session_false_when_no_db(self): |
| 104 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 105 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 106 | + adapter = WhatsAppWebAdapter(creds_dir=tmpdir) |
| 107 | + assert adapter.has_saved_session() is False |
| 108 | + |
| 109 | + def test_has_saved_session_true_when_db_exists(self): |
| 110 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 111 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 112 | + adapter = WhatsAppWebAdapter(creds_dir=tmpdir, bot_name="test") |
| 113 | + # Create a fake db file |
| 114 | + db_path = os.path.join(tmpdir, "test.db") |
| 115 | + with open(db_path, "w") as f: |
| 116 | + f.write("fake-db-content") |
| 117 | + assert adapter.has_saved_session() is True |
| 118 | + |
| 119 | + @pytest.mark.asyncio |
| 120 | + async def test_logout_removes_db_files(self): |
| 121 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 122 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 123 | + adapter = WhatsAppWebAdapter(creds_dir=tmpdir, bot_name="test") |
| 124 | + # Create fake db files |
| 125 | + for suffix in [".db", ".db-wal", ".db-shm"]: |
| 126 | + path = os.path.join(tmpdir, f"test{suffix}") |
| 127 | + with open(path, "w") as f: |
| 128 | + f.write("x") |
| 129 | + assert adapter.has_saved_session() is True |
| 130 | + await adapter.logout() |
| 131 | + assert adapter.has_saved_session() is False |
| 132 | + assert not os.path.exists(os.path.join(tmpdir, "test.db-wal")) |
| 133 | + assert not os.path.exists(os.path.join(tmpdir, "test.db-shm")) |
| 134 | + |
| 135 | + |
| 136 | +# ── JID normalization ───────────────────────────────────────────── |
| 137 | + |
| 138 | +class TestJIDNormalization: |
| 139 | + """Test phone number to JID conversion.""" |
| 140 | + |
| 141 | + def test_already_jid(self): |
| 142 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 143 | + assert WhatsAppWebAdapter._normalize_jid("1234@s.whatsapp.net") == "1234@s.whatsapp.net" |
| 144 | + |
| 145 | + def test_phone_number(self): |
| 146 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 147 | + assert WhatsAppWebAdapter._normalize_jid("15551234567") == "15551234567@s.whatsapp.net" |
| 148 | + |
| 149 | + def test_phone_with_plus(self): |
| 150 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 151 | + assert WhatsAppWebAdapter._normalize_jid("+15551234567") == "15551234567@s.whatsapp.net" |
| 152 | + |
| 153 | + def test_group_jid_passthrough(self): |
| 154 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 155 | + assert WhatsAppWebAdapter._normalize_jid("123456@g.us") == "123456@g.us" |
| 156 | + |
| 157 | + def test_phone_with_spaces_and_dashes(self): |
| 158 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 159 | + assert WhatsAppWebAdapter._normalize_jid("+1 555-123-4567") == "15551234567@s.whatsapp.net" |
| 160 | + |
| 161 | + |
| 162 | +# ── Connect / Disconnect ────────────────────────────────────────── |
| 163 | + |
| 164 | +class TestConnectDisconnect: |
| 165 | + """Test connect and disconnect lifecycle (mocked neonize).""" |
| 166 | + |
| 167 | + @pytest.mark.asyncio |
| 168 | + async def test_connect_raises_without_neonize(self): |
| 169 | + """connect() raises ImportError when neonize is not installed.""" |
| 170 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 171 | + adapter = WhatsAppWebAdapter(creds_dir="/tmp/test-no-neonize") |
| 172 | + |
| 173 | + with patch.dict(sys.modules, { |
| 174 | + "neonize": None, |
| 175 | + "neonize.aioze": None, |
| 176 | + "neonize.aioze.client": None, |
| 177 | + "neonize.aioze.events": None, |
| 178 | + }): |
| 179 | + with pytest.raises(ImportError, match="neonize"): |
| 180 | + await adapter.connect() |
| 181 | + |
| 182 | + @pytest.mark.asyncio |
| 183 | + async def test_disconnect_when_not_connected(self): |
| 184 | + """disconnect() is safe to call when not connected.""" |
| 185 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 186 | + adapter = WhatsAppWebAdapter() |
| 187 | + await adapter.disconnect() |
| 188 | + assert adapter.is_connected is False |
| 189 | + |
| 190 | + @pytest.mark.asyncio |
| 191 | + async def test_disconnect_calls_client_disconnect(self): |
| 192 | + """disconnect() calls client.disconnect() if client exists.""" |
| 193 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 194 | + adapter = WhatsAppWebAdapter() |
| 195 | + mock_client = AsyncMock() |
| 196 | + adapter._client = mock_client |
| 197 | + adapter._is_connected = True |
| 198 | + |
| 199 | + await adapter.disconnect() |
| 200 | + mock_client.disconnect.assert_called_once() |
| 201 | + assert adapter.is_connected is False |
| 202 | + assert adapter._client is None |
| 203 | + |
| 204 | + |
| 205 | +# ── Send message ────────────────────────────────────────────────── |
| 206 | + |
| 207 | +class TestSendMessage: |
| 208 | + """Test send_message with mocked client.""" |
| 209 | + |
| 210 | + @pytest.mark.asyncio |
| 211 | + async def test_send_when_not_connected_returns_none(self): |
| 212 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 213 | + adapter = WhatsAppWebAdapter() |
| 214 | + result = await adapter.send_message("1234@s.whatsapp.net", "hello") |
| 215 | + assert result is None |
| 216 | + |
| 217 | + @pytest.mark.asyncio |
| 218 | + async def test_send_reaction_when_not_connected(self): |
| 219 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 220 | + adapter = WhatsAppWebAdapter() |
| 221 | + # Should not raise |
| 222 | + await adapter.send_reaction("1234@s.whatsapp.net", "msg123", "👍") |
| 223 | + |
| 224 | + @pytest.mark.asyncio |
| 225 | + async def test_mark_as_read_when_not_connected(self): |
| 226 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 227 | + adapter = WhatsAppWebAdapter() |
| 228 | + # Should not raise |
| 229 | + await adapter.mark_as_read("1234@s.whatsapp.net", "msg123") |
| 230 | + |
| 231 | + |
| 232 | +# ── QR code display ─────────────────────────────────────────────── |
| 233 | + |
| 234 | +class TestQRCodeDisplay: |
| 235 | + """Test QR code terminal display.""" |
| 236 | + |
| 237 | + def test_print_qr_with_qrcode_lib(self): |
| 238 | + """Uses qrcode library when available.""" |
| 239 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 240 | + mock_qr = MagicMock() |
| 241 | + mock_qr_module = MagicMock() |
| 242 | + mock_qr_module.QRCode.return_value = mock_qr |
| 243 | + |
| 244 | + with patch.dict(sys.modules, {"qrcode": mock_qr_module}): |
| 245 | + WhatsAppWebAdapter._print_qr_terminal("test-qr-data") |
| 246 | + mock_qr_module.QRCode.assert_called_once() |
| 247 | + mock_qr.add_data.assert_called_once_with("test-qr-data") |
| 248 | + |
| 249 | + def test_print_qr_fallback_without_lib(self, capsys): |
| 250 | + """Falls back to text output when qrcode not installed.""" |
| 251 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 252 | + with patch.dict(sys.modules, {"qrcode": None}): |
| 253 | + # Force ImportError |
| 254 | + original_import = __builtins__.__import__ if hasattr(__builtins__, '__import__') else __import__ |
| 255 | + def mock_import(name, *args, **kwargs): |
| 256 | + if name == "qrcode": |
| 257 | + raise ImportError("No module named 'qrcode'") |
| 258 | + return original_import(name, *args, **kwargs) |
| 259 | + |
| 260 | + with patch("builtins.__import__", side_effect=mock_import): |
| 261 | + WhatsAppWebAdapter._print_qr_terminal("test-qr-data-fallback") |
| 262 | + captured = capsys.readouterr() |
| 263 | + assert "QR" in captured.out or "qrcode" in captured.out.lower() |
| 264 | + |
| 265 | + |
| 266 | +# ── Safe callback ───────────────────────────────────────────────── |
| 267 | + |
| 268 | +class TestSafeCallback: |
| 269 | + """Test the _safe_callback helper.""" |
| 270 | + |
| 271 | + @pytest.mark.asyncio |
| 272 | + async def test_sync_callback(self): |
| 273 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 274 | + adapter = WhatsAppWebAdapter() |
| 275 | + called = [] |
| 276 | + def cb(x): |
| 277 | + called.append(x) |
| 278 | + await adapter._safe_callback(cb, "test") |
| 279 | + assert called == ["test"] |
| 280 | + |
| 281 | + @pytest.mark.asyncio |
| 282 | + async def test_async_callback(self): |
| 283 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 284 | + adapter = WhatsAppWebAdapter() |
| 285 | + called = [] |
| 286 | + async def cb(x): |
| 287 | + called.append(x) |
| 288 | + await adapter._safe_callback(cb, "async-test") |
| 289 | + assert called == ["async-test"] |
| 290 | + |
| 291 | + @pytest.mark.asyncio |
| 292 | + async def test_callback_error_does_not_raise(self): |
| 293 | + from praisonai.bots._whatsapp_web_adapter import WhatsAppWebAdapter |
| 294 | + adapter = WhatsAppWebAdapter() |
| 295 | + def cb(): |
| 296 | + raise ValueError("boom") |
| 297 | + # Should not raise |
| 298 | + await adapter._safe_callback(cb) |
0 commit comments