Jack1808/Claude_Code

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