Method test_dashboard_bridge_wired

tests/agents/test_loop.py:203–228

Loop wires dashboard bridge input queue and broadcasts messages.

(self)

    @pytest.mark.asyncio
    async def test_dashboard_bridge_wired(self):
        """Loop wires dashboard bridge input queue and broadcasts messages."""
        ctx = _make_context()
        bridge = MagicMock()
        bridge.set_input_queue = MagicMock()
        bridge.on_message = AsyncMock()
        ctx.dashboard_bridge = bridge

        ui = _make_ui()
        q: asyncio.Queue = asyncio.Queue()
        done = asyncio.Event()

        await q.put("hello")
        await q.put("exit")

        mock_convo = MagicMock()
        mock_convo.process_conversation = AsyncMock()

        with patch(
            "mcp_cli.chat.conversation.ConversationProcessor",
            return_value=mock_convo,
        ):
            await run_agent_loop(ctx, ui, q, done)

        bridge.set_input_queue.assert_called_once_with(q)
        bridge.on_message.assert_called_once_with("user", "hello")

    @pytest.mark.asyncio
    async def test_max_response_length_truncated(self):