"""The window: a whole turn through the real MainWindow, worker thread and Bridge against the loopback desk server from test_console — only the mouse is faked (by clicking the approval button programmatically).""" import json import os import sys import threading import time 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.QtWidgets import QApplication, QLabel from bolt_console.gui import widgets from bolt_console.gui.bridge import Bridge from bolt_console.gui.window import MainWindow, make_app from tests.test_console import desk # noqa: F401 (fixture) app = make_app(["test"]) def pump_until(predicate, timeout=10.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: app.processEvents() if predicate(): return True time.sleep(0.01) return False def entries(window, kind): layout = window.entries return [layout.itemAt(i).widget() for i in range(layout.count()) if isinstance(layout.itemAt(i).widget(), kind)] @pytest.fixture def window(desk, monkeypatch): # noqa: F811 from bolt_console import config monkeypatch.setattr(config, "STREAMING_REPLIES", True) win = MainWindow(mode="ask") yield win win.close() def test_confirm_blocks_the_worker_until_the_window_answers(): bridge = Bridge() asked = [] bridge.approval_needed.connect(lambda tool_id, q: asked.append(q)) result = {} worker = threading.Thread(target=lambda: result.setdefault("a", bridge.confirm("Run it?"))) worker.start() assert pump_until(lambda: asked == ["Run it?"]) assert worker.is_alive() # still waiting on the user bridge.answer("always") worker.join(2) assert result["a"] == "always" def test_close_releases_a_pending_approval_as_no(): bridge = Bridge() result = {} worker = threading.Thread(target=lambda: result.setdefault("a", bridge.confirm("Run it?"))) worker.start() time.sleep(0.05) bridge.close() worker.join(2) assert result["a"] == "no" assert bridge.confirm("again?") == "no" # and never blocks after close def test_a_turn_with_an_approved_edit(window, desk, tmp_path): # noqa: F811 target = tmp_path / "app.py" target.write_text("DEBUG = True\n", encoding="utf-8") edit = "filectl " + json.dumps( {"op": "edit", "path": str(target), "old": "DEBUG = True", "new": "DEBUG = False"}) desk.script["/desk/converse_stream"] = [ {"type": "say", "text": "On it."}, {"type": "command", "command": edit, "token": "t1"}, ] desk.script["/desk/tool_result"] = {"type": "reply", "text": "Debug is **off** now."} window._send("turn debug off") assert pump_until(lambda: entries(window, widgets.ToolCard) and entries(window, widgets.ToolCard)[0].approval.isVisibleTo(window)) card = entries(window, widgets.ToolCard)[0] assert card.pill.property("kind") == "waiting" assert "DEBUG = True" in card.diff.text() card.allow_button.click() assert pump_until(lambda: not window._busy) assert target.read_text(encoding="utf-8") == "DEBUG = False\n" assert card.pill.property("kind") == "ok" texts = [m.text() for m in entries(window, widgets.AssistantMessage)] assert texts == ["On it.", "Debug is **off** now."] assert len(entries(window, widgets.UserMessage)) == 1 def test_deny_reports_back_and_the_turn_finishes(window, desk, tmp_path): # noqa: F811 desk.script["/desk/converse"] = {"type": "command", "command": "echo hi > made.txt", "token": "t"} desk.script["/desk/tool_result"] = {"type": "reply", "text": "Okay, I won't."} window._send("make a file") assert pump_until(lambda: entries(window, widgets.ToolCard) and entries(window, widgets.ToolCard)[0].approval.isVisibleTo(window)) card = entries(window, widgets.ToolCard)[0] card._answer("no") assert pump_until(lambda: not window._busy) assert not (tmp_path / "made.txt").exists() assert card.pill.property("kind") == "declined" posted = [r[2] for r in desk.requests if r[1] == "/desk/tool_result"] assert "declined" in posted[0]["output"] def test_always_allow_flips_the_header_switch_to_auto(window, desk, tmp_path): # noqa: F811 desk.script["/desk/converse"] = {"type": "command", "command": "echo hi", "token": "t"} window._send("say hi") assert pump_until(lambda: entries(window, widgets.ToolCard) and entries(window, widgets.ToolCard)[0].approval.isVisibleTo(window)) entries(window, widgets.ToolCard)[0]._answer("always") assert pump_until(lambda: not window._busy) assert window.session.tools.mode == "auto" assert window.mode.buttons["auto"].isChecked() def test_clear_empties_the_transcript_but_keeps_the_scaffolding(window, desk): # noqa: F811 window._send("hello") assert pump_until(lambda: not window._busy) window.clear_transcript() app.processEvents() assert entries(window, widgets.UserMessage) == [] assert window.entries.indexOf(window.thinking) != -1 assert not window.empty.isHidden() def test_markdown_code_uses_the_app_mono_font(): html = widgets.markdown_html("run `ls` now", QLabel().font()) assert "Courier New" not in html assert widgets.theme.CODE in html def test_align_columns_pads_tab_separated_rows(): rows = widgets.align_columns(["dir/", "a.txt\t5B", "longer.txt\t12B"]) assert rows[0] == "dir/" assert rows[1].index("5B") == rows[2].index("12B") def test_new_chat_clears_the_window_and_starts_a_fresh_session(window, desk, tmp_path, monkeypatch): # noqa: F811 from bolt_console import config monkeypatch.setattr(config, "STATE_FILE", tmp_path / "session.json") desk.script["/desk/chats"] = {"session_id": "chat-3f9a2b1c9e04"} # POST mints the id window._send("hello") assert pump_until(lambda: not window._busy) window.start_new_chat() assert pump_until(lambda: config.SESSION_ID == "chat-3f9a2b1c9e04") assert ("POST", "/desk/chats", {}) in desk.requests # over real HTTP, off the UI thread assert entries(window, widgets.UserMessage) == [] assert not window.empty.isHidden() window._send("next") assert pump_until(lambda: not window._busy) converse = [r[2] for r in desk.requests if r[0] == "POST" and "converse" in r[1]] assert converse[-1]["session_id"] == "chat-3f9a2b1c9e04" # the next turn uses the new chat def test_write_ico_holds_every_size_the_exe_needs(tmp_path): from PySide6.QtGui import QImageReader from bolt_console.gui import icons path = tmp_path / "bolt.ico" icons.write_ico(str(path)) reader = QImageReader(str(path)) sizes = [] while True: image = reader.read() if image.isNull(): break sizes.append(image.width()) if not reader.jumpToNextImage(): break assert sizes == list(icons.APP_ICON_SIZES)