rpa_vision_v3/tests/unit/test_chat_window_paused_dispatch.py

"""Tests pour ChatWindow._dispatch_paused_action.

Couvre le routage bus SocketIO → fallback HTTP de la bulle paused.
Le bug d'origine ``paused_bubble: bus déconnecté, resume non émis``
était causé par l'absence de ce fallback (cf.
``docs/CR_AUDIT_PAUSED_RESUME_BUS_2026-05-22.md``).

Les tests appellent ``ChatWindow._dispatch_paused_action`` en tant
que fonction unbound avec un faux ``self`` (``SimpleNamespace``) pour
éviter de démarrer Tkinter pendant les tests unitaires.
"""

from __future__ import annotations

import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock

ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(ROOT))

from agent_v0.agent_v1.ui.chat_window import ChatWindow  # noqa: E402


def _make_self(bus=None, server_client=None):
    return SimpleNamespace(_bus=bus, _server_client=server_client)


def _call(mock_self, replay_id="replay_xyz",
          bus_method="resume_replay", client_method="resume_replay"):
    return ChatWindow._dispatch_paused_action(
        mock_self, replay_id, bus_method=bus_method, client_method=client_method,
    )


class TestDispatchPausedAction:
    def test_bus_connected_and_emits_uses_bus(self):
        bus = MagicMock(connected=True)
        bus.resume_replay.return_value = True
        client = MagicMock(resume_replay=MagicMock(return_value=True))
        emitted, channel = _call(_make_self(bus=bus, server_client=client))
        assert emitted is True
        assert channel == "bus"
        bus.resume_replay.assert_called_once_with("replay_xyz")
        client.resume_replay.assert_not_called()

    def test_bus_disconnected_falls_back_to_http(self):
        bus = MagicMock(connected=False)
        client = MagicMock(resume_replay=MagicMock(return_value=True))
        emitted, channel = _call(_make_self(bus=bus, server_client=client))
        assert emitted is True
        assert channel == "http"
        bus.resume_replay.assert_not_called()
        client.resume_replay.assert_called_once_with("replay_xyz")

    def test_bus_emit_returns_false_falls_back_to_http(self):
        """Bus marqué connecté mais l'emit retourne False (socket cassé
        entre connect() et send) → bascule sur HTTP."""
        bus = MagicMock(connected=True)
        bus.resume_replay.return_value = False
        client = MagicMock(resume_replay=MagicMock(return_value=True))
        emitted, channel = _call(_make_self(bus=bus, server_client=client))
        assert emitted is True
        assert channel == "http"

    def test_bus_emit_raises_falls_back_to_http(self):
        bus = MagicMock(connected=True)
        bus.resume_replay.side_effect = RuntimeError("socket broken")
        client = MagicMock(resume_replay=MagicMock(return_value=True))
        emitted, channel = _call(_make_self(bus=bus, server_client=client))
        assert emitted is True
        assert channel == "http"

    def test_no_bus_uses_http_directly(self):
        client = MagicMock(resume_replay=MagicMock(return_value=True))
        emitted, channel = _call(_make_self(bus=None, server_client=client))
        assert emitted is True
        assert channel == "http"

    def test_all_channels_fail_returns_false(self):
        """Cas critique : bus déconnecté ET HTTP injoignable → l'UI
        doit ré-activer les boutons côté appelant. Ici on vérifie
        juste que dispatch retourne (False, '')."""
        bus = MagicMock(connected=False)
        client = MagicMock(resume_replay=MagicMock(return_value=False))
        emitted, channel = _call(_make_self(bus=bus, server_client=client))
        assert emitted is False
        assert channel == ""

    def test_neither_bus_nor_client_returns_false(self):
        emitted, channel = _call(_make_self(bus=None, server_client=None))
        assert emitted is False
        assert channel == ""

    def test_client_method_missing_falls_through(self):
        """Si server_client est un vieux client sans resume_replay,
        on ne plante pas — on retourne (False, '')."""
        bus = MagicMock(connected=False)
        legacy_client = SimpleNamespace()  # pas de resume_replay
        emitted, channel = _call(
            _make_self(bus=bus, server_client=legacy_client),
        )
        assert emitted is False
        assert channel == ""

    def test_abort_routing_symmetric(self):
        """Le même mécanisme couvre l'abort — vérifie qu'on utilise
        bien la méthode demandée par le caller."""
        bus = MagicMock(connected=False)
        client = MagicMock(abort_replay=MagicMock(return_value=True))
        emitted, channel = _call(
            _make_self(bus=bus, server_client=client),
            bus_method="abort_replay",
            client_method="abort_replay",
        )
        assert emitted is True
        assert channel == "http"
        client.abort_replay.assert_called_once_with("replay_xyz")


class TestPausedBubbleHeight:
    """Couvre _compute_paused_bubble_height — patch troncature 22 mai 2026."""

    def test_empty_message_uses_minimum_height(self):
        h, scroll = ChatWindow._compute_paused_bubble_height("")
        assert h == 2
        assert scroll is False

    def test_short_message_no_scrollbar(self):
        h, scroll = ChatWindow._compute_paused_bubble_height("Court message.")
        assert h == 2
        assert scroll is False

    def test_long_single_line_triggers_scrollbar(self):
        # ~600 chars sans \n → wrapped_lines = 600 // 60 + 1 = 11
        msg = "x" * 600
        h, scroll = ChatWindow._compute_paused_bubble_height(msg)
        assert h == 11
        assert scroll is True

    def test_message_with_many_newlines_uses_explicit_count(self):
        """Cas du bug : reason serveur listant 6 candidats sur 6 lignes
        courtes — wrapped_lines bas mais explicit_lines élevé."""
        msg = "\n".join([f"option {i}" for i in range(6)])
        h, scroll = ChatWindow._compute_paused_bubble_height(msg)
        # 6 lignes explicites > 2 lignes wrappées → hauteur = 6
        assert h == 6
        # Pas encore au cap, contenu court → pas de scrollbar
        assert scroll is False

    def test_cap_reached_triggers_scrollbar_even_if_short(self):
        """Quand on dépasse le cap (12 lignes), la scrollbar DOIT
        s'afficher quel que soit la longueur en caractères."""
        msg = "\n".join([f"l{i}" for i in range(20)])
        h, scroll = ChatWindow._compute_paused_bubble_height(msg)
        assert h == 12  # plafond
        assert scroll is True

    def test_long_content_triggers_scrollbar_at_200_chars(self):
        """Seuil sécurité texte : ≥ 200 chars → scrollbar même si
        peu de lignes (filet anti-troncature visuel)."""
        msg = "x" * 220
        h, scroll = ChatWindow._compute_paused_bubble_height(msg)
        assert scroll is True