File size: 6,190 Bytes
6172a47 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """Tests for messaging/ module."""
import json
from unittest.mock import patch
import pytest
# --- Existing Tests ---
class TestMessagingModels:
"""Test messaging models."""
def test_incoming_message_creation(self):
"""Test IncomingMessage dataclass."""
from messaging.models import IncomingMessage
msg = IncomingMessage(
text="Hello",
chat_id="123",
user_id="456",
message_id="789",
platform="telegram",
)
assert msg.text == "Hello"
assert msg.chat_id == "123"
assert msg.platform == "telegram"
assert msg.is_reply() is False
def test_incoming_message_with_reply(self):
"""Test IncomingMessage as a reply."""
from messaging.models import IncomingMessage
msg = IncomingMessage(
text="Reply text",
chat_id="123",
user_id="456",
message_id="789",
platform="discord",
reply_to_message_id="100",
)
assert msg.is_reply() is True
assert msg.reply_to_message_id == "100"
class TestMessagingBase:
"""Test MessagingPlatform ABC."""
def test_platform_is_abstract(self):
"""Verify MessagingPlatform cannot be instantiated."""
from messaging.platforms.base import MessagingPlatform
with pytest.raises(TypeError):
MessagingPlatform()
class TestSessionStore:
"""Test SessionStore."""
def test_session_store_init(self, tmp_path):
"""Test SessionStore initialization."""
from messaging.session import SessionStore
store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
assert store._trees == {}
# --- Tree Tests ---
def test_save_and_get_tree(self, tmp_path):
"""Test saving and retrieving trees."""
from messaging.session import SessionStore
store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
tree_data = {
"root": "r1",
"nodes": {"r1": {"content": "root"}, "n1": {"content": "child"}},
}
store.save_tree("r1", tree_data)
loaded = store.get_tree("r1")
assert loaded == tree_data
# Verify node mapping
node_map = store.get_node_mapping()
assert node_map["r1"] == "r1"
assert node_map["n1"] == "r1"
def test_register_node(self, tmp_path):
"""Test manual node registration."""
from messaging.session import SessionStore
store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
store.register_node("n_manual", "r_manual")
assert store.get_node_mapping()["n_manual"] == "r_manual"
# --- Persistence & Edge Cases ---
def test_load_existing_file_with_trees(self, tmp_path):
"""Test loading file with trees (legacy sessions ignored)."""
from messaging.session import SessionStore
data = {
"sessions": {},
"trees": {"r1": {"root_id": "r1", "nodes": {"r1": {}}}},
"node_to_tree": {"r1": "r1"},
"message_log": {},
}
p = tmp_path / "sessions.json"
with open(p, "w") as f:
json.dump(data, f)
store = SessionStore(storage_path=str(p))
assert store.get_tree("r1") is not None
def test_load_corrupt_file(self, tmp_path):
"""Test loading corrupt/invalid json file."""
p = tmp_path / "sessions.json"
with open(p, "w") as f:
f.write("{invalid json")
from messaging.session import SessionStore
# Should log error and start empty, avoiding crash
store = SessionStore(storage_path=str(p))
assert store._trees == {}
def test_save_error_handling(self, tmp_path):
"""Test error during save."""
from messaging.session import SessionStore
store = SessionStore(storage_path=str(tmp_path / "sessions.json"))
store.save_tree("r1", {"root_id": "r1", "nodes": {"r1": {}}})
# Mock open to raise exception
with patch("builtins.open", side_effect=OSError("Disk full")):
store.save_tree("r2", {"root_id": "r2", "nodes": {"r2": {}}})
# Should log error but not crash. Tree should be in memory.
assert "r2" in store._trees
class TestTreeQueueManager:
"""Test TreeQueueManager."""
def test_tree_queue_manager_init(self):
"""Test TreeQueueManager initialization."""
from messaging.trees.queue_manager import TreeQueueManager
mgr = TreeQueueManager()
assert mgr.get_tree_count() == 0
def test_tree_not_busy_initially(self):
"""Test tree is not busy when no messages."""
from messaging.trees.queue_manager import TreeQueueManager
mgr = TreeQueueManager()
assert mgr.is_tree_busy("nonexistent") is False
def test_get_queue_size_empty(self):
"""Test queue size is 0 for non-existent node."""
from messaging.trees.queue_manager import TreeQueueManager
mgr = TreeQueueManager()
assert mgr.get_queue_size("nonexistent") == 0
@pytest.mark.asyncio
async def test_create_tree_and_enqueue(self):
"""Test creating a tree and enqueueing."""
from messaging.models import IncomingMessage
from messaging.trees.queue_manager import TreeQueueManager
mgr = TreeQueueManager()
processed = []
async def processor(node_id, node):
processed.append(node_id)
incoming = IncomingMessage(
text="test", chat_id="1", user_id="1", message_id="1", platform="test"
)
await mgr.create_tree("1", incoming, "status_1")
was_queued = await mgr.enqueue("1", processor)
# First message should process immediately, not queue
assert was_queued is False
@pytest.mark.asyncio
async def test_cancel_tree_empty(self):
"""Test cancelling non-existent tree."""
from messaging.trees.queue_manager import TreeQueueManager
mgr = TreeQueueManager()
cancelled = await mgr.cancel_tree("nonexistent")
assert cancelled == []
|