Bolt-Console/tests/test_chats.py

"""The Chats dialog, the header New chat, /chats in the terminal, and the
row helpers. No network: list/create are injected fakes, and they still run
on the real QThreadPool, so the off-UI-thread hand-off is exercised."""

import io
import os
import sys
import threading
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest
from PySide6.QtCore import Qt

from bolt_console import chats, config, console, server_client
from bolt_console.tools import Tools
from bolt_console.ui import Terminal

CHATS = [
    {"session_id": "chat-new1", "label": "Fix the build",
     "last_message": "try again [Bolt Console: a text terminal on Windows 10; working directory C:/x. To act",
     "last_active": "2026-09-25T23:46:31+00:00"},
    {"session_id": "pet-console-test", "label": "console DESKTOP",
     "last_message": "what is in this folder", "last_active": "2026-09-25T21:26:20+00:00"},
    {"session_id": "android-SM-S928U", "label": "android SM S928U",
     "last_message": "wifi networks", "last_active": "2026-09-20T22:33:34+00:00"},
]


@pytest.fixture(autouse=True)
def _session(monkeypatch, tmp_path):
    monkeypatch.setattr(config, "SESSION_ID", "pet-console-test")
    monkeypatch.setattr(config, "STATE_FILE", tmp_path / "session.json")
    monkeypatch.setattr(config, "SERVER_URL", "http://127.0.0.1:9")
    monkeypatch.setattr(config, "API_KEY", "k")
    monkeypatch.delenv("DESK_SESSION_ID", raising=False)


class Calls:
    """Fake list/create that record which thread they ran on."""

    def __init__(self, chat_list=CHATS, new_id="chat-fresh", fail=None):
        self.chat_list, self.new_id, self.fail, self.threads = chat_list, new_id, fail, []

    def list(self):
        self.threads.append(threading.current_thread())
        if self.fail:
            raise server_client.ServerError(self.fail)
        return self.chat_list

    def create(self):
        self.threads.append(threading.current_thread())
        if self.fail:
            raise server_client.ServerError(self.fail)
        return self.new_id


# ── helpers ─────────────────────────────────────────────────────────────────

def test_preview_drops_client_context_notes_even_when_cut_off():
    assert chats.clean_preview(CHATS[0]["last_message"]) == "try again"
    assert chats.clean_preview("hi [on screen right now: VS Code] there") == "hi there"
    assert chats.clean_preview("x" * 200, 20).endswith("…")


def test_relative_time():
    now = datetime(2026, 9, 26, 0, 0, tzinfo=timezone.utc)

    def ago(**delta):
        return (now - timedelta(**delta)).isoformat()

    assert chats.relative_time(ago(seconds=10), now) == "just now"
    assert chats.relative_time(ago(minutes=12), now) == "12 min ago"
    assert chats.relative_time(ago(hours=3), now) == "3 h ago"
    assert chats.relative_time(ago(days=1, hours=2), now) == "yesterday"
    assert chats.relative_time(ago(days=4), now) == "4 days ago"
    assert chats.relative_time("garbage", now) == "garbage"


# ── terminal ────────────────────────────────────────────────────────────────

def terminal_session(calls):
    out = io.StringIO()
    ui = Terminal(out=out, color=False, input_fn=lambda _p: "", animate=False)
    session = console.Session(ui, tools=Tools(ui), list_chats=calls.list,
                              create_chat=calls.create, list_files=lambda: [])
    return session, out


def test_chats_command_lists_and_marks_the_current_one():
    session, out = terminal_session(Calls())
    session.handle("/chats")
    text = out.getvalue()
    assert "Fix the build" in text and "try again" in text and "Bolt Console:" not in text
    assert "*  2  console DESKTOP" in text  # the current chat is starred


def test_chat_command_switches_by_number_and_remembers_it():
    session, _ = terminal_session(Calls())
    session.handle("/chats")
    session.handle("/chat 3")
    assert config.SESSION_ID == "android-SM-S928U"
    assert config._remembered_session() == "android-SM-S928U"


def test_new_uses_the_server_minted_id_and_falls_back_locally():
    session, _ = terminal_session(Calls(new_id="chat-abc123"))
    session.handle("/new")
    assert config.SESSION_ID == "chat-abc123"
    session, out = terminal_session(Calls(fail="HTTP 401"))
    session.handle("/new")
    assert config.SESSION_ID.startswith("pet-console-") and "made locally" in out.getvalue()


# ── window ──────────────────────────────────────────────────────────────────

def pump(app, predicate, timeout=5.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        app.processEvents()
        if predicate():
            return True
        time.sleep(0.01)
    return False


@pytest.fixture
def window():
    from bolt_console.gui.window import MainWindow, make_app

    app = make_app(["test"])
    calls = Calls()
    win = MainWindow(list_chats=calls.list, create_chat=calls.create)
    win.calls, win.app = calls, app
    yield win
    win.close()


def test_dialog_lists_chats_off_the_ui_thread(window):
    window.open_chats()
    dialog = window.chat_dialog
    assert pump(window.app, lambda: dialog.list.count() == 3)
    assert window.calls.threads and all(t is not threading.main_thread() for t in window.calls.threads)
    assert dialog.list.currentItem().data(Qt.UserRole) == "pet-console-test"  # current one selected
    dialog.close()


def test_picking_a_chat_switches_session_and_clears_the_transcript(window):
    from bolt_console.gui.widgets import UserMessage

    window._add(UserMessage("old conversation"))
    window.open_chats()
    dialog = window.chat_dialog
    assert pump(window.app, lambda: dialog.list.count() == 3)
    dialog.list.setCurrentRow(0)
    dialog.open_button.click()
    window.app.processEvents()
    assert config.SESSION_ID == "chat-new1"
    assert not window.empty.isHidden()
    assert "Fix the build" in window.hint.text()


def test_new_chat_in_the_dialog_switches_to_the_minted_id(window):
    window.open_chats()
    dialog = window.chat_dialog
    assert pump(window.app, lambda: dialog.list.count() == 3)
    dialog.new_button.click()
    assert pump(window.app, lambda: config.SESSION_ID == "chat-fresh")
    assert not dialog.isVisible()


def test_header_new_chat_mints_off_thread_and_clears(window):
    from bolt_console.gui.widgets import UserMessage

    window._add(UserMessage("old"))
    window.start_new_chat()
    assert pump(window.app, lambda: config.SESSION_ID == "chat-fresh")
    assert window.calls.threads[0] is not threading.main_thread()
    assert not window.empty.isHidden() and window.new_chat.isEnabled()


def test_header_new_chat_failure_keeps_the_current_chat(window):
    window.calls.fail = "HTTP 401 unauthorized"
    window.start_new_chat()
    assert pump(window.app, lambda: "Couldn't start a new chat" in window.hint.text())
    assert config.SESSION_ID == "pet-console-test"
    assert window.new_chat.isEnabled()


def test_dialog_shows_a_load_failure(window):
    window.calls.fail = "HTTP 401"
    window.open_chats()
    dialog = window.chat_dialog
    assert pump(window.app, lambda: "Couldn't load chats" in dialog.status.text())
    assert dialog.refresh_button.isEnabled()
    dialog.close()


def test_switching_is_refused_while_bolt_is_working(window):
    window._set_busy(True)
    window.open_chats()
    assert "Wait for Bolt" in window.hint.text()
    window.switch_chat("chat-other", "Other")
    assert config.SESSION_ID == "pet-console-test"
    window._set_busy(False)