path: root/test_llamachat.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-only
#
# llamachat - a small native chat client for a local llama.cpp router
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
"""Self-checks for the non-GUI logic. Run: ./test_llamachat.py"""

import json
import os
import sys
import tempfile
from pathlib import Path

# Some checks need PySide6, so reuse the launcher's venv discovery rather
# than requiring the venv interpreter to be named on the command line.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import llamachat_venv  # noqa: E402

llamachat_venv.reexec(__file__)

from llamachat import backend, config, db

PRESETS_SAMPLE = """\
version = 1

; A preset whose mmproj line is commented out -> not vision capable.
#[Disabled-Model]
#ngl = all
#mmproj = /models/disabled/mmproj.gguf

[vision-model]
ngl = all
ctx-size = 32768
mmproj = /models/vision/mmproj.gguf

[text-model]
ngl = all
ctx-size = 16384
; --- Multimodal ---
; WARNING: eats VRAM.
#mmproj = /models/text/mmproj.gguf

[no-ctx-model]
ngl = all
"""


def test_presets():
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "presets.ini"
        path.write_text(PRESETS_SAMPLE)
        presets = config.parse_presets(path)

        # A '#'-commented section must not appear at all.
        assert "Disabled-Model" not in presets, presets.keys()
        assert set(presets) == {"vision-model", "text-model", "no-ctx-model"}

        # mmproj present -> vision; commented out -> not vision.
        assert presets["vision-model"].vision is True
        assert presets["text-model"].vision is False
        assert presets["no-ctx-model"].vision is False

        assert presets["vision-model"].ctx_size == 32768
        assert presets["text-model"].ctx_size == 16384
        assert presets["no-ctx-model"].ctx_size == 4096  # default

        # 32768 tokens * 3.5 chars * 0.5 -> 57344 chars
        assert presets["vision-model"].char_budget(0.5, 3.5) == 57344

    # A missing file yields no presets rather than raising.
    assert config.parse_presets(Path("/nonexistent/presets.ini")) == {}
    print("ok  presets parsing")


def test_real_presets():
    """The user's actual file, if present, must classify as expected."""
    path = Path("/etc/llama-server/presets.ini")
    if not path.exists():
        print("skip real presets (file absent)")
        return
    presets = config.parse_presets(path)
    assert presets, "presets.ini exists but parsed to nothing"

    # Section names track whatever the user currently runs, so assert the
    # parsing properties rather than a list of names that goes stale on
    # every rename.
    for preset in presets.values():
        assert preset.name
        assert preset.ctx_size > 0
        assert isinstance(preset.vision, bool)

    # Whether a given section has vision is the user's choice and changes
    # when they edit the file; only the commented-out case is a parsing
    # claim, and PRESETS_SAMPLE covers that hermetically above.
    print(f"ok  real presets classification ({len(presets)} sections)")


def test_fts_query_escaping():
    # Bare punctuation and FTS keywords must not become query syntax.
    assert db._fts_query("hello world") == '"hello" "world"'
    assert db._fts_query("foo AND bar") == '"foo" "AND" "bar"'
    assert db._fts_query('say "hi"') == '"say" """hi"""'
    assert db._fts_query("-flag") == '"-flag"'
    assert db._fts_query("   ") == ""
    print("ok  fts query escaping")


def test_history_roundtrip():
    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "test.db")

        chat = history.create_session("chat", "vision-model", "About otters")
        history.add_message(chat, "user", "Tell me about otters please")
        history.add_message(chat, "assistant", "Otters are semiaquatic mammals")

        shot = history.create_session("oneshot", "text-model", "Capital city")
        history.add_message(shot, "user", "What is the capital of Italy")

        rows = history.messages(chat)
        assert len(rows) == 2
        assert rows[0]["role"] == "user"

        # Newest session first.
        recent = history.recent_sessions()
        assert len(recent) == 2
        assert recent[0]["id"] == shot

        # FTS finds content across sessions, and punctuation cannot break it.
        assert len(history.search("otters")) == 2
        assert len(history.search("capital")) == 1
        assert history.search("zebra") == []
        assert history.search('otters "AND') == []  # must not raise

        # Attachments survive with enough detail to reconstruct context.
        message_id = history.add_message(chat, "user", "look at this")
        history.add_attachment(
            message_id, "/home/u/pic.png", "image",
            mime="image/png", size=1234, sha256="abc", thumb=b"\xff\xd8jpeg",
        )
        saved = history.attachments(message_id)
        assert len(saved) == 1
        assert saved[0]["path"] == "/home/u/pic.png"
        assert saved[0]["kind"] == "image"
        assert saved[0]["thumb"] == b"\xff\xd8jpeg"

        # Editing a streamed message keeps the FTS index in step.
        streamed = history.add_message(chat, "assistant", "")
        history.update_message(streamed, "the platypus is unusual")
        assert len(history.search("platypus")) == 1

        # Deleting a session takes its messages out of search too.
        history.delete_session(chat)
        assert history.search("otters") == []
        assert len(history.recent_sessions()) == 1

        history.close()
    print("ok  history roundtrip")


def test_attachment_truncation():
    with tempfile.TemporaryDirectory() as tmp:
        big = Path(tmp) / "big.py"
        big.write_text("x" * 5000)

        att = backend.load_attachment(big, char_budget=1000)
        assert att.kind == "text"
        assert att.truncated is True
        assert len(att.text) == 1000
        assert att.size == 5000
        assert len(att.sha256) == 64

        small = Path(tmp) / "small.txt"
        small.write_text("hello")
        att = backend.load_attachment(small, char_budget=1000)
        assert att.truncated is False
        assert att.text == "hello"

        odd = Path(tmp) / "thing.bin"
        odd.write_bytes(b"\x00\x01")
        try:
            backend.load_attachment(odd, char_budget=1000)
        except backend.BackendError:
            pass
        else:
            raise AssertionError("unsupported type should raise")
    print("ok  attachment truncation")


def test_classify():
    assert backend.classify(Path("a.py")) == "text"
    assert backend.classify(Path("a.SlackBuild")) == "text"
    assert backend.classify(Path("a.png")) == "image"
    assert backend.classify(Path("a.jpg")) == "image"
    assert backend.classify(Path("a.so")) == "unknown"
    print("ok  file classification")


def test_sse_parsing():
    line = 'data: {"choices":[{"delta":{"content":"hi"}}]}'
    assert backend._parse_sse_line(line) == ("content", "hi")

    # Reasoning arrives in its own field, which is what lets the UI keep
    # thinking and reply apart without parsing <think> tags.
    think = 'data: {"choices":[{"delta":{"reasoning_content":"hmm"}}]}'
    assert backend._parse_sse_line(think) == ("reasoning", "hmm")

    assert backend._parse_sse_line("data: [DONE]") == backend.DONE
    assert backend._parse_sse_line(": keepalive") is None
    assert backend._parse_sse_line("") is None
    assert backend._parse_sse_line("data: {bad json") is None

    # An opening delta of {'role': 'assistant', 'content': None} carries
    # nothing to render and must not be mistaken for end-of-stream.
    opening = backend._parse_sse_line(
        'data: {"choices":[{"delta":{"role":"assistant","content":null}}]}'
    )
    assert opening is None
    assert opening != backend.DONE

    # A delta carrying both fields is thinking, not reply. Classifying it as
    # content splices the tail of the reasoning onto the front of the reply.
    both = backend._parse_sse_line(
        'data: {"choices":[{"delta":'
        '{"reasoning_content":"still thinking","content":""}}]}'
    )
    assert both == ("reasoning", "still thinking"), both

    # An empty reasoning delta renders nothing and must not fall through to
    # the content branch.
    empty = backend._parse_sse_line(
        'data: {"choices":[{"delta":{"reasoning_content":""}}]}'
    )
    assert empty is None, empty
    print("ok  sse parsing")


def test_reasoning_storage():
    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "r.db")
        sid = history.create_session("chat", "m", "t")

        # A streamed reply is inserted empty, then filled in once done.
        mid = history.add_message(sid, "assistant", "")
        history.update_message(mid, "51", "the model's private thinking")

        row = history.messages(sid)[0]
        assert row["content"] == "51"
        assert row["reasoning"] == "the model's private thinking"

        # Reasoning must stay out of the search index, or thinking text
        # would drown out real hits.
        assert history.search("51") != []
        assert history.search("private") == []

        # Omitting the argument leaves stored reasoning untouched.
        history.update_message(mid, "52")
        assert history.messages(sid)[0]["reasoning"] == (
            "the model's private thinking"
        )
        history.close()
    print("ok  reasoning storage")


def test_migration_adds_reasoning():
    """A database created before the reasoning column must still open."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'user','older message',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        rows = history.messages(1)
        assert rows[0]["content"] == "older message"
        assert rows[0]["reasoning"] == ""  # backfilled by the migration

        mid = history.add_message(1, "assistant", "new")
        history.update_message(mid, "new", "fresh thinking")
        assert history.messages(1)[1]["reasoning"] == "fresh thinking"
        history.close()
    print("ok  reasoning column migration")


def test_user_content():
    with tempfile.TemporaryDirectory() as tmp:
        src = Path(tmp) / "code.py"
        src.write_text("print(1)")
        text_att = backend.load_attachment(src, 1000)

        # Text only -> a plain string, file inlined before the prompt.
        content = backend.build_user_content("explain", [text_att])
        assert isinstance(content, str)
        assert "print(1)" in content
        assert content.endswith("explain")

        # No attachments -> just the prompt.
        assert backend.build_user_content("hi", []) == "hi"

        # An image -> the multi-part array the vision API expects.
        img = backend.Attachment(
            path=Path("/x/a.png"), kind="image", mime="image/png",
            size=1, sha256="", data_url="data:image/png;base64,AAA",
        )
        content = backend.build_user_content("what is this", [img])
        assert isinstance(content, list)
        assert content[0]["type"] == "text"
        assert content[1]["type"] == "image_url"
        assert content[1]["image_url"]["url"].startswith("data:image/png")
    print("ok  user content assembly")


def test_markdown_rendering():
    """Replies render as markdown; markup inside them stays literal."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    fragment = _markdown_to_fragment(
        "**bold** and *italic* and `code`\n\n"
        "- a\n- b\n\n"
        "| x | y |\n|---|---|\n| 1 | 2 |\n\n"
        "```py\ndef f():\n    pass\n```\n"
    )
    assert "font-weight:700" in fragment
    assert "font-style:italic" in fragment
    assert "<ul" in fragment
    assert "<table" in fragment
    assert "<pre" in fragment
    # Every <pre> is tinted, since Qt emits one per line of a code block.
    assert fragment.count("<pre") == fragment.count("background:rgba")
    # The document wrapper must not leak into the fragment.
    assert "<html" not in fragment and "<body" not in fragment

    # Markup in a reply is shown, never interpreted.
    hostile = _markdown_to_fragment(
        'Text <b>tag</b> <img src=x onerror=alert(1)> <a href="http://bad">l</a>'
    )
    assert "&lt;b&gt;" in hostile, hostile
    assert "font-weight:700" not in hostile
    # The URL may appear, but only as escaped text, never as a live anchor.
    assert "<a href" not in hostile
    assert "&lt;a href=&quot;http://bad&quot;&gt;" in hostile, hostile

    # A markdown link is a real anchor, so check a reply cannot forge one
    # aimed at the reasoning toggle.
    from llamachat.ui import REASONING_SCHEME

    forged = _markdown_to_fragment(f"[click]({REASONING_SCHEME}0)")
    assert f'href="{REASONING_SCHEME}' not in forged, forged
    assert 'href="blocked:' in forged, forged

    # Ordinary markdown links still work.
    link = _markdown_to_fragment("[site](http://example.com)")
    assert "http://example.com" in link

    # Empty input renders nothing rather than an empty document wrapper.
    assert _markdown_to_fragment("") == ""

    # Partial markdown arrives constantly while streaming and must not raise.
    for partial in ("```", "```py", "```py\ndef f(", "| a |", "**unclosed", "#"):
        _markdown_to_fragment(partial)
    print("ok  markdown rendering")


def test_unbalanced_backticks():
    """An odd backtick cannot reflow the rest of a reply as code."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _close_fences, _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    # An unclosed fence gets one, so the tail is not swallowed.
    assert _close_fences("a\n```py\nx=1").endswith("\n```")
    # A closed fence is left exactly as it was.
    balanced = "a\n```py\nx=1\n```\ntail"
    assert _close_fences(balanced) == balanced
    # A longer opener needs a closer at least as long.
    assert _close_fences("````\nx").endswith("\n````")
    # Tildes are fences too.
    assert _close_fences("~~~\nx").endswith("\n~~~")
    # A lone inline tick is dropped rather than opening a run.
    assert "`" not in _close_fences("use ` here")
    # Balanced inline ticks survive untouched.
    assert _close_fences("a `b` c") == "a `b` c"

    # The real failure: reasoning text full of ticks must not become one
    # block that eats the prose after it.
    leaked = "` (wait)\nthen `find / -perm -4000` and more prose here"
    rendered = _markdown_to_fragment(leaked)
    assert "more prose here" in rendered
    assert "<pre" not in rendered, rendered
    print("ok  unbalanced backticks")


def test_system_qt_theme_guard():
    """The plugin path is only borrowed when the Qt versions agree."""
    import os

    from llamachat.__main__ import _system_qt_version, _use_system_qt_theme

    # An explicit user setting is never overridden.
    saved = os.environ.get("QT_PLUGIN_PATH")
    try:
        os.environ["QT_PLUGIN_PATH"] = "/user/choice"
        _use_system_qt_theme()
        assert os.environ["QT_PLUGIN_PATH"] == "/user/choice"
    finally:
        if saved is None:
            os.environ.pop("QT_PLUGIN_PATH", None)
        else:
            os.environ["QT_PLUGIN_PATH"] = saved

    system = _system_qt_version()
    if not system:
        print("skip qt theme guard (no system Qt6 found)")
        return
    assert system.count(".") == 2, system

    from PySide6.QtCore import qVersion

    saved = os.environ.pop("QT_PLUGIN_PATH", None)
    try:
        _use_system_qt_theme()
        chosen = os.environ.get("QT_PLUGIN_PATH")
        if system == qVersion():
            assert chosen, "matching Qt versions should adopt system plugins"
            assert Path(chosen, "platformthemes").is_dir()
        else:
            assert chosen is None, (
                f"must not load {system} plugins into Qt {qVersion()}"
            )
    finally:
        if saved is None:
            os.environ.pop("QT_PLUGIN_PATH", None)
        else:
            os.environ["QT_PLUGIN_PATH"] = saved
    print("ok  system qt theme guard")


def test_prompt_store():
    """Prompts are files; a preset replaces the global one."""
    from llamachat import prompts

    with tempfile.TemporaryDirectory() as tmp:
        store = prompts.PromptStore(Path(tmp) / "prompts")
        assert store.names() == []

        store.ensure_default()
        assert store.names() == ["default"]
        assert store.path_for("default").exists()

        # ensure_default must not clobber an edited global prompt.
        store.save("default", "edited global")
        store.ensure_default()
        assert store.load("default").text == "edited global"

        store.save("coding", "you write code")
        store.save("terse", "be brief")
        # Global first, then presets alphabetically.
        assert store.names() == ["default", "coding", "terse"]

        # A preset replaces the global prompt rather than adding to it.
        assert store.resolve("coding") == "you write code"
        assert store.resolve("") == "edited global"
        assert store.resolve("default") == "edited global"

        # Sentinels.
        assert store.resolve(prompts.NONE) == ""
        assert store.resolve(prompts.CUSTOM, "one off") == "one off"

        # A preset deleted behind our back falls back rather than sending
        # nothing, which would silently change the model's behaviour.
        assert store.resolve("missing") == "edited global"

        assert store.delete("coding") is True
        assert store.names() == ["default", "terse"]
        # The global prompt is the fallback for everything, so it stays.
        assert store.delete("default") is False
        assert "default" in store.names()

        # Names become filenames, so path separators must not escape.
        assert prompts.safe_name("../../etc/passwd") == "etc-passwd"
        assert prompts.safe_name("/etc/passwd") == "etc-passwd"
        assert prompts.safe_name("my prompt!") == "my-prompt"
        for hostile in ("../escape", "a/b", "/tmp/x"):
            saved = store.save(hostile, "x")
            assert saved.path.parent == store.dir, saved.path
            assert saved.path.resolve().parent == store.dir.resolve()

        # A name that reduces to nothing is rejected rather than writing
        # a dotfile called ".md".
        for empty in ("..", "", "....", "///"):
            assert prompts.safe_name(empty) == ""
            assert store.load(empty) is None
            assert store.delete(empty) is False
            try:
                store.path_for(empty)
            except ValueError:
                pass
            else:
                raise AssertionError(f"{empty!r} should be rejected")
    print("ok  prompt store")


def test_token_estimate():
    """The meter's pre-send estimate tracks message size."""
    small = [{"role": "user", "content": "hi"}]
    large = [{"role": "user", "content": "word " * 1000}]

    assert backend.estimate_tokens([], 3.5) == 0
    assert backend.estimate_tokens(small, 3.5) < 20
    assert backend.estimate_tokens(large, 3.5) > 1000

    # More history means a bigger prompt.
    grown = large + [{"role": "assistant", "content": "reply " * 500}]
    assert backend.estimate_tokens(grown, 3.5) > backend.estimate_tokens(
        large, 3.5
    )

    # An image costs far more than its text part suggests.
    with_image = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "what is this"},
                {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAA"}},
            ],
        }
    ]
    assert backend.estimate_tokens(with_image, 3.5) > 500

    # A silly ratio must not divide by zero.
    assert backend.estimate_tokens(small, 0) > 0
    print("ok  token estimate")


def test_usage_parsing():
    """The final stream chunk carries the exact prompt token count."""
    line = (
        'data: {"choices":[],"usage":{"prompt_tokens":1234,'
        '"completion_tokens":56,"total_tokens":1290}}'
    )
    kind, payload = backend._parse_sse_line(line)
    assert kind == "usage"
    stats = json.loads(payload)
    assert stats["prompt_tokens"] == 1234
    assert stats["total_tokens"] == 1290

    # A usage-less chunk with empty choices is not mistaken for one.
    assert backend._parse_sse_line('data: {"choices":[]}') is None
    print("ok  usage parsing")


def test_session_prompt_storage():
    """A conversation remembers the prompt it was built with."""
    from llamachat import prompts

    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "p.db")

        sid = history.create_session(
            "chat", "m", "t", prompt_name="coding", prompt_custom=""
        )
        row = history.get_session(sid)
        assert row["prompt_name"] == "coding"
        assert row["prompt_custom"] == ""

        history.set_prompt(sid, prompts.CUSTOM, "just this once")
        row = history.get_session(sid)
        assert row["prompt_name"] == prompts.CUSTOM
        assert row["prompt_custom"] == "just this once"

        # Defaults keep older call sites working.
        plain = history.create_session("oneshot", "m", "t")
        assert history.get_session(plain)["prompt_name"] == ""
        history.close()
    print("ok  session prompt storage")


def test_prompt_column_migration():
    """A database predating the prompt columns must still open."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        row = history.get_session(1)
        assert row["title"] == "old"
        assert row["prompt_name"] == ""
        assert row["prompt_custom"] == ""
        history.set_prompt(1, "coding")
        assert history.get_session(1)["prompt_name"] == "coding"
        history.close()
    print("ok  prompt column migration")


def test_sidebar_toggle():
    """The history panel hides, restores its width, and persists."""
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat import backend as _backend
    from llamachat import config as _config
    from llamachat.ui import ChatWindow

    app = QApplication.instance() or QApplication([])
    assert app is not None

    with tempfile.TemporaryDirectory() as tmp:
        cfg = _config.load(Path(tmp) / "config.toml")
        cfg.prompts_dir = Path(tmp) / "prompts"
        cfg.db_path = Path(tmp) / "t.db"
        # state_path already points inside tmp, since it is derived from the
        # config path, so these checks cannot touch the real saved layout.
        assert cfg.state_path.parent == Path(tmp), cfg.state_path

        history = db.History(cfg.db_path)
        client = _backend.Client(cfg.base_url)
        presets = _config.parse_presets(cfg.presets_path)

        window = ChatWindow(cfg, history, client, presets)
        window.resize(1000, 700)

        # The shortcut must be registered on the window itself.
        shortcuts = [
            a.shortcut().toString()
            for a in window.actions()
            if not a.shortcut().isEmpty()
        ]
        assert "Ctrl+\\" in shortcuts, shortcuts

        window.show()
        assert window.sidebar_visible()
        assert window.sidebar_button.isChecked()

        window.toggle_sidebar()
        assert not window.sidebar_visible()
        assert not window.sidebar_button.isChecked()

        window.toggle_sidebar()
        assert window.sidebar_visible()

        # The width survives a hide/show rather than snapping to a default.
        window.splitter.setSizes([333, 667])
        app.processEvents()
        window.toggle_sidebar()
        window.toggle_sidebar()
        assert window.sidebar_width == 333, window.sidebar_width

        # Driving the button directly must take the same path.
        window.sidebar_button.setChecked(False)
        app.processEvents()
        assert not window.sidebar_visible()
        window.sidebar_button.setChecked(True)
        app.processEvents()
        assert window.sidebar_visible()
        assert window.sidebar_width == 333, window.sidebar_width

        # Hidden state and width must come back on the next start.
        window.sidebar_button.setChecked(False)
        window._save_layout()
        restored = ChatWindow(cfg, history, client, presets)
        assert not restored.sidebar_button.isChecked()
        assert restored.sidebar_width == 333, restored.sidebar_width

        restored.close()
        window.close()
        history.close()
    print("ok  sidebar toggle")


def test_shortcuts():
    """Ctrl+N, Ctrl+F and Escape behave as advertised."""
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtCore import QEvent, Qt
    from PySide6.QtGui import QKeyEvent
    from PySide6.QtWidgets import QApplication

    from llamachat import backend as _backend
    from llamachat import config as _config
    from llamachat.ui import ChatWindow

    app = QApplication.instance() or QApplication([])

    with tempfile.TemporaryDirectory() as tmp:
        cfg = _config.load(Path(tmp) / "config.toml")
        cfg.prompts_dir = Path(tmp) / "prompts"
        cfg.db_path = Path(tmp) / "t.db"
        history = db.History(cfg.db_path)
        window = ChatWindow(
            cfg,
            history,
            _backend.Client(cfg.base_url),
            _config.parse_presets(cfg.presets_path),
        )
        window.resize(1000, 700)
        window.show()
        window.raise_()
        window.activateWindow()
        # The offscreen platform only grants focus to the active window, and
        # a window left over from an earlier check can still hold it.
        window.setFocus()
        app.processEvents()

        registered = {
            a.shortcut().toString()
            for a in window.actions()
            if not a.shortcut().isEmpty()
        }
        for wanted in ("Ctrl+N", "Ctrl+F", "Ctrl+\\"):
            assert wanted in registered, (wanted, registered)

        # Ctrl+F puts the cursor in the search box. The offscreen platform
        # only grants real focus to one window per process, so check where
        # focus was directed rather than whether the platform granted it.
        window.input.setFocus()
        app.processEvents()
        window.focus_search()
        app.processEvents()
        assert window.focusWidget() is window.search_box, window.focusWidget()

        # The search box lives in the top bar, so hiding the history panel
        # must not take it away.
        window.sidebar_button.setChecked(False)
        app.processEvents()
        assert not window.sidebar_visible()
        assert window.search_box.isVisible()

        # Its results render in the panel, so focusing reveals the panel.
        window.focus_search()
        app.processEvents()
        assert window.sidebar_visible()

        def press_escape() -> None:
            window.keyPressEvent(
                QKeyEvent(QEvent.KeyPress, Qt.Key_Escape, Qt.NoModifier)
            )
            app.processEvents()

        # Escape backs out of the search box before hiding the window, so a
        # stray press while filtering does not dismiss everything.
        window.search_box.setFocus()
        window.search_box.setText("otters")
        app.processEvents()
        assert window.focusWidget() is window.search_box
        press_escape()
        assert window.search_box.text() == ""
        assert window.isVisible()
        press_escape()
        assert window.focusWidget() is window.input
        assert window.isVisible()
        press_escape()
        assert not window.isVisible()

        # Ctrl+N clears the conversation and puts the cursor in the input.
        window.show()
        window.raise_()
        window.activateWindow()
        # The offscreen platform only grants focus to the active window, and
        # a window left over from an earlier check can still hold it.
        window.setFocus()
        app.processEvents()
        session = history.create_session("chat", "m", "old")
        history.add_message(session, "user", "something")
        window.open_session(session)
        assert window.session_id == session
        window.search_box.setFocus()
        window.new_session()
        app.processEvents()
        assert window.session_id is None
        assert window.transcript.toPlainText().strip() == ""
        assert window.focusWidget() is window.input

        window.close()
        history.close()
    print("ok  shortcuts")


def test_version_matches_changelog():
    """The package version must be the newest release in the changelog."""
    import re

    import llamachat

    assert re.fullmatch(r"\d+\.\d+\.\d+", llamachat.__version__), (
        f"not semver: {llamachat.__version__}"
    )

    changelog = Path(__file__).resolve().parent / "CHANGELOG.md"
    if not changelog.exists():
        print("skip changelog check (file absent)")
        return

    released = re.findall(
        r"^## \[(\d+\.\d+\.\d+)\]", changelog.read_text(), re.MULTILINE
    )
    assert released, "changelog has no released versions"
    assert released[0] == llamachat.__version__, (
        f"__version__ is {llamachat.__version__} but the newest changelog "
        f"entry is {released[0]}"
    )
    print("ok  version matches changelog")


def test_venv_discovery():
    """The launcher must find a venv even when it shares the system binary."""
    saved = os.environ.pop("LLAMACHAT_PYTHON", None)
    try:
        found = llamachat_venv.find_interpreter()
        current = Path(sys.executable)
        other = [
            c for c in llamachat_venv.CANDIDATES if c.is_file() and c != current
        ]
        if other:
            # A venv built with --system-site-packages has a bin/python3 that
            # symlinks to the system interpreter. Comparing resolved paths
            # would discard it as "the interpreter we are already running".
            assert found is not None, (
                f"a candidate exists ({other[0]}) but none was selected"
            )
            assert found.is_file()
        else:
            # Already running as the only candidate; nothing to hand over to.
            assert found is None

        # An explicit override wins over the search order.
        os.environ["LLAMACHAT_PYTHON"] = sys.executable
        chosen = llamachat_venv.find_interpreter()
        # Only rejected because it is the interpreter already running.
        assert chosen is None or chosen == Path(sys.executable)

        os.environ["LLAMACHAT_PYTHON"] = "/nonexistent/python3"
        assert llamachat_venv.find_interpreter() is None
    finally:
        if saved is None:
            os.environ.pop("LLAMACHAT_PYTHON", None)
        else:
            os.environ["LLAMACHAT_PYTHON"] = saved

    # reexec must be a no-op once PySide6 is importable, or it would loop.
    if llamachat_venv.have_pyside():
        llamachat_venv.reexec(__file__)  # returns rather than exec'ing
    print("ok  venv discovery")


def test_config_defaults():
    cfg = config.load(Path("/nonexistent/config.toml"))
    assert cfg.base_url == "http://localhost:8181"
    assert cfg.presets_path == Path("/etc/llama-server/presets.ini")
    assert cfg.socket_path.name == "llamachat.sock"
    assert cfg.db_path.name == "history.db"

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "config.toml"
        path.write_text(
            'base_url = "http://localhost:9999/"\ndefault_model = "m"\n'
        )
        cfg = config.load(path)
        assert cfg.base_url == "http://localhost:9999"  # trailing / stripped
        assert cfg.default_model == "m"
        assert cfg.request_timeout == 300  # default kept
    print("ok  config defaults")


class _FakeResponse:
    """Enough of an http.client response for urlopen's context manager."""

    def __init__(self, body: bytes):
        self._body = body

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def _fake_urlopen(body, capture=None):
    """A urlopen replacement returning `body`, or raising it when an error."""

    def opener(request, timeout=None):
        if capture is not None:
            capture.append(request.full_url)
        if isinstance(body, Exception):
            raise body
        return _FakeResponse(body)

    return opener


def _with_urlopen(body, capture=None):
    """Swap search's urlopen for a fake. Returns the original to restore."""
    import urllib.request

    original = urllib.request.urlopen
    urllib.request.urlopen = _fake_urlopen(body, capture)
    return original


def test_search_tool_schema():
    from llamachat import search

    schema = search.TOOL_SCHEMA
    assert schema["type"] == "function"
    function = schema["function"]
    assert function["name"] == "web_search"
    assert function["description"]
    params = function["parameters"]
    assert params["type"] == "object"
    assert params["required"] == ["query"]
    assert params["properties"]["query"]["type"] == "string"

    # Disabled search must not put a tools key on the wire at all, or a
    # model that ignores it still pays for the tokens.
    sent = []
    client = backend.Client("http://x")
    client._stream_once = lambda model, messages, tools: (
        sent.append(tools) or iter([("content", "hi")])
    )
    out = list(client.stream_chat("m", [], backend.SearchConfig(enabled=False)))
    assert out == [("content", "hi")]
    assert sent == [None], sent

    # One search per turn by default: a follow-up call arrives as literal
    # <tool_call> XML rather than a structured call, and a turn that ends on
    # one shows the user an empty reply.
    assert config.DEFAULTS["max_searches"] == 1
    assert config.load(Path("/nonexistent/config.toml")).max_searches == 1

    # A configured-but-urlless setup resolves to disabled at config load.
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "config.toml"
        path.write_text('search_enabled = true\nsearch_url = ""\n')
        assert config.load(path).search_enabled is False

        path.write_text(
            'search_enabled = true\nsearch_url = "http://searx.local:8888/"\n'
        )
        cfg = config.load(path)
        assert cfg.search_enabled is True
        assert cfg.search_url == "http://searx.local:8888"  # trailing / gone
    print("ok  search tool schema")


def test_search_results_sanitising():
    import urllib.request

    from llamachat import search
    from llamachat.ui import SEARCH_SCHEME, _markdown_to_fragment, _search_html

    payload = json.dumps(
        {
            "results": [
                {
                    "title": "First",
                    "url": "https://example.com/a",
                    "content": "x" * 500,
                    # Fields the model has no business seeing.
                    "engine": "duckduckgo",
                    "score": 1.5,
                    "positions": [1],
                },
                {"title": "Second", "url": "https://example.com/b", "content": "s"},
                {"title": "Third", "url": "https://example.com/c", "content": "t"},
            ]
        }
    ).encode()

    original = _with_urlopen(payload)
    try:
        results = search.search("http://searx", "q", count=2, snippet_chars=100)
    finally:
        urllib.request.urlopen = original

    # count caps the list; only three fields survive; snippets truncate.
    assert len(results) == 2, results
    assert set(results[0]) == {"title", "url", "content"}, results[0]
    assert len(results[0]["content"]) == 100
    assert "engine" not in results[0]

    # Result text is escaped on display, so markup in a snippet stays text.
    hostile = [
        {
            "query": "q",
            "error": "",
            "results": [
                {
                    "title": "<script>alert(1)</script>",
                    "url": "https://example.com/x",
                    "content": f"click [here]({SEARCH_SCHEME}0)",
                }
            ],
        }
    ]
    rendered = _search_html(hostile, 0, expanded=True)
    assert "<script>" not in rendered, rendered
    assert "&lt;script&gt;" in rendered
    # The scheme appears exactly once, as this block's own toggle. The copy
    # inside the snippet stayed literal text rather than becoming an href.
    assert rendered.count(f'href="{SEARCH_SCHEME}') == 1, rendered

    # A reply forging the scheme as a markdown link gets it defused.
    forged = _markdown_to_fragment(f"[expand]({SEARCH_SCHEME}0)")
    assert SEARCH_SCHEME not in forged, forged
    assert "blocked:" in forged
    print("ok  search result sanitising")


def test_tool_call_accumulation():
    from llamachat import search

    calls: dict = {}
    # The id and name arrive first, the arguments in pieces after it.
    search.accumulate(calls, [
        {"index": 0, "id": "call_1", "function": {"name": "web_search", "arguments": ""}}
    ])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": '{"que'}}])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": 'ry": "kern'}}])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": 'el"}'}}])

    assert list(calls) == [0]
    assert calls[0]["id"] == "call_1"
    assert calls[0]["name"] == "web_search"
    assert search.parse_query(calls[0]["arguments"]) == "kernel"

    # Two calls in one round stay apart, keyed by index.
    pair: dict = {}
    search.accumulate(pair, [
        {"index": 0, "id": "a", "function": {"name": "web_search", "arguments": '{"query":"one"}'}},
        {"index": 1, "id": "b", "function": {"name": "web_search", "arguments": '{"query":"two"}'}},
    ])
    assert search.parse_query(pair[0]["arguments"]) == "one"
    assert search.parse_query(pair[1]["arguments"]) == "two"

    # Unusable arguments yield no query rather than an exception.
    assert search.parse_query("{not json") == ""
    assert search.parse_query("[]") == ""
    assert search.parse_query('{"query": 7}') == ""
    assert search.parse_query("") == ""
    print("ok  tool call accumulation")


def _tool_round(query: str = "q"):
    """One streamed round that asks for a search."""
    return [
        (
            "tool_calls",
            json.dumps(
                [
                    {
                        "index": 0,
                        "id": "call_1",
                        "function": {
                            "name": "web_search",
                            "arguments": json.dumps({"query": query}),
                        },
                    }
                ]
            ),
        ),
        ("tool_finish", ""),
    ]


def test_search_loop_cap():
    import urllib.request

    from llamachat import search

    sent_tools = []
    client = backend.Client("http://x")

    def always_tool_calls(model, messages, tools):
        sent_tools.append(tools)
        yield from _tool_round()

    client._stream_once = always_tool_calls
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        cfg = backend.SearchConfig(
            enabled=True, url="http://searx", max_searches=2
        )
        list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    # max_searches rounds offer the tool, then one final round without it.
    assert len(sent_tools) == 3, sent_tools
    assert sent_tools[0] == [search.TOOL_SCHEMA]
    assert sent_tools[1] == [search.TOOL_SCHEMA]
    assert sent_tools[-1] is None, "the final round must withdraw the tool"

    # A round that never asks for a tool ends the turn immediately.
    quiet = []

    def no_tools(model, messages, tools):
        quiet.append(tools)
        yield ("content", "done")

    client._stream_once = no_tools
    out = list(client.stream_chat("m", [], backend.SearchConfig(
        enabled=True, url="http://searx")))
    assert out == [("content", "done")]
    assert len(quiet) == 1, quiet
    print("ok  search loop cap")


def test_search_failure_paths():
    import urllib.error
    import urllib.request

    from llamachat import search

    cases = [
        (TimeoutError("timed out"), "Cannot reach SearXNG"),
        (urllib.error.URLError("Connection refused"), "Cannot reach SearXNG"),
        (b"<html>not json</html>", "did not return JSON"),
    ]
    for body, expected in cases:
        original = _with_urlopen(body)
        try:
            failed = ""
            try:
                search.search("http://searx", "q")
            except search.SearchError as exc:
                failed = str(exc)
        finally:
            urllib.request.urlopen = original
        assert expected in failed, (body, failed)

    # Zero results with healthy engines is a success with an empty list:
    # the web really had nothing.
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        assert search.search("http://searx", "q") == []
    finally:
        urllib.request.urlopen = original

    # Zero results *because* every engine was rate-limited or CAPTCHA'd is
    # a failed search. Reporting it as "no results" would tell the model
    # the web is empty and invite an answer from stale training data.
    dead = json.dumps(
        {
            "results": [],
            "unresponsive_engines": [
                ["duckduckgo", "CAPTCHA"],
                ["brave", "Suspended: too many requests"],
            ],
        }
    ).encode()
    original = _with_urlopen(dead)
    try:
        failed = ""
        try:
            search.search("http://searx", "q")
        except search.SearchError as exc:
            failed = str(exc)
    finally:
        urllib.request.urlopen = original
    assert "every search engine failed" in failed, failed
    assert "duckduckgo: CAPTCHA" in failed, failed

    # Engines that failed while others still answered are not an error:
    # partial results are results.
    partial = json.dumps(
        {
            "results": [{"title": "T", "url": "https://e.com", "content": "c"}],
            "unresponsive_engines": [["brave", "timeout"]],
        }
    ).encode()
    original = _with_urlopen(partial)
    try:
        assert len(search.search("http://searx", "q")) == 1
    finally:
        urllib.request.urlopen = original

    # A malformed unresponsive_engines field must not crash the summary.
    assert search._unresponsive({"unresponsive_engines": "nonsense"}) == ""
    assert search._unresponsive({"unresponsive_engines": [["solo"]]}) == "solo"
    assert search._unresponsive({}) == ""

    # An unconfigured URL fails before any request is attempted.
    try:
        search.search("", "q")
        raise AssertionError("empty url must raise")
    except search.SearchError:
        pass

    # Every failure still completes the turn and still tells the model.
    client = backend.Client("http://x")
    rounds = [_tool_round("kernel"), [("content", "answered anyway")]]
    seen_messages = []

    def scripted(model, messages, tools):
        seen_messages.append([dict(m) for m in messages])
        yield from rounds[min(len(seen_messages) - 1, len(rounds) - 1)]

    client._stream_once = scripted
    original = _with_urlopen(urllib.error.URLError("Connection refused"))
    try:
        cfg = backend.SearchConfig(enabled=True, url="http://searx")
        out = list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    assert ("content", "answered anyway") in out, out
    starts = [p for k, p in out if k == "search_start"]
    dones = [json.loads(p) for k, p in out if k == "search_done"]
    assert starts == ["kernel"], starts
    assert dones and dones[0]["error"], dones
    assert dones[0]["results"] == []

    # The second request carries the assistant tool call and a tool reply
    # naming the failure, so the model knows the search did not happen.
    second = seen_messages[1]
    assert second[-2]["role"] == "assistant"
    assert second[-2]["tool_calls"][0]["id"] == "call_1"
    tool_msg = second[-1]
    assert tool_msg["role"] == "tool"
    assert tool_msg["tool_call_id"] == "call_1"
    assert search.RESULT_PREFIX in tool_msg["content"]
    assert "error" in json.loads(
        tool_msg["content"][len(search.RESULT_PREFIX):].strip()
    )

    # A malformed tool call is answered rather than left dangling, or the
    # next request would be rejected for an unanswered call.
    bad = {"id": "call_9", "name": "web_search", "arguments": "{not json"}
    convo: list = []
    emitted = list(client._run_search(convo, bad, backend.SearchConfig(
        enabled=True, url="http://searx")))
    assert emitted == [], emitted  # nothing searched, nothing displayed
    assert convo[-1]["role"] == "tool"
    assert convo[-1]["tool_call_id"] == "call_9"
    print("ok  search failure paths")


def test_search_storage():
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "s.db")
        sid = history.create_session("chat", "m", "t")
        mid = history.add_message(sid, "assistant", "")

        records = [
            {
                "query": "latest kernel",
                "results": [
                    {
                        "title": "Kernel",
                        "url": "https://kernel.org",
                        "content": "snippet",
                    }
                ],
                "error": "",
            }
        ]
        history.update_message(mid, "7.1.5", "thinking", json.dumps(records))

        row = history.messages(sid)[0]
        assert row["content"] == "7.1.5"
        assert row["reasoning"] == "thinking"
        assert json.loads(row["searches"]) == records

        # Snippets stay out of the FTS index: web text the user never wrote
        # must not compete with their own messages.
        assert history.search("7.1.5") != []
        assert history.search("snippet") == []

        # Omitting the argument leaves stored searches untouched.
        history.update_message(mid, "7.1.6")
        assert json.loads(history.messages(sid)[0]["searches"]) == records

        # A message with no searches stores NULL rather than an empty list.
        plain = history.add_message(sid, "assistant", "")
        history.update_message(plain, "no search", "", None)
        assert history.messages(sid)[1]["searches"] is None
        history.close()

    # A database predating the column must still open and read as no
    # searches rather than raising.
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'assistant','older',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        from llamachat.ui import _column, _searches

        row = history.messages(1)[0]
        assert row["searches"] is None
        assert _searches(_column(row, "searches")) == []
        history.close()
    print("ok  search storage")


def test_date_note():
    import datetime

    from llamachat import ui
    from llamachat.ui import _date_note

    fixed = datetime.datetime(2026, 7, 31, 14, 30)
    note = _date_note(fixed)

    # The real date must be stated, or the model falls back on its cutoff.
    assert "Friday, 31 July 2026" in note, note
    # And it must be told not to date its own queries, which is the bug
    # that poisons results before they are even fetched.
    assert "year in a search query" in note, note

    # Composed onto the chosen prompt rather than replacing it, so a preset
    # keeps its instructions.
    class _Cfg:
        search_enabled = True

    class _Win:
        cfg = _Cfg()
        system_prompt_text = staticmethod(lambda: "Be terse.")
        _system_messages = ui.ChatWindow._system_messages

    msgs = _Win._system_messages(_Win())
    assert len(msgs) == 1
    assert msgs[0]["role"] == "system"
    assert msgs[0]["content"].startswith("Be terse.")
    assert "2026" in msgs[0]["content"]

    # With no system prompt selected, the date still goes: the model has no
    # clock either way.
    class _Bare(_Win):
        system_prompt_text = staticmethod(lambda: "")

    bare = _Bare._system_messages(_Bare())
    assert len(bare) == 1
    assert "Today's date is" in bare[0]["content"]

    # Search off means no date note and no system message at all.
    class _Off(_Bare):
        class cfg:
            search_enabled = False

    assert _Off._system_messages(_Off()) == []
    print("ok  date note")


def test_search_html():
    from llamachat.ui import SEARCH_SCHEME, _search_html, _searches

    assert _search_html([], 0, False) == ""

    ok = [
        {
            "query": "latest kernel",
            "error": "",
            "results": [
                {
                    "title": "Kernel.org",
                    "url": "https://kernel.org",
                    "content": "The Linux Kernel Archives",
                },
                {
                    "title": "Wikipedia",
                    "url": "https://en.wikipedia.org/wiki/Linux",
                    "content": "An operating system kernel",
                },
            ],
        }
    ]

    collapsed = _search_html(ok, 3, expanded=False)
    assert "▸" in collapsed
    assert "searched: latest kernel (2 results)" in collapsed
    assert f'href="{SEARCH_SCHEME}3"' in collapsed
    # Collapsed shows the summary only, never the sources.
    assert "kernel.org" not in collapsed

    expanded = _search_html(ok, 3, expanded=True)
    assert "▾" in expanded
    assert "Kernel.org" in expanded
    assert 'href="https://kernel.org"' in expanded
    assert "The Linux Kernel Archives" in expanded

    # A failure says so, and names the reason.
    failed = _search_html(
        [{"query": "kernel", "results": [], "error": "Cannot reach SearXNG"}],
        0,
        expanded=False,
    )
    assert "search failed: kernel" in failed
    assert "Cannot reach SearXNG" in failed

    # A non-web URL is shown but never becomes a clickable anchor.
    sneaky = _search_html(
        [
            {
                "query": "q",
                "error": "",
                "results": [
                    {"title": "T", "url": "file:///etc/passwd", "content": "c"}
                ],
            }
        ],
        0,
        expanded=True,
    )
    assert 'href="file:' not in sneaky, sneaky
    assert "file:///etc/passwd" in sneaky

    # The stored column decodes back into what rendering expects, and junk
    # in that column degrades to no block rather than raising.
    assert _searches(json.dumps(ok)) == ok
    assert _searches("") == []
    assert _searches("{not json") == []
    assert _searches('{"query": "not a list"}') == []
    print("ok  search html")


if __name__ == "__main__":
    test_presets()
    test_real_presets()
    test_fts_query_escaping()
    test_history_roundtrip()
    test_attachment_truncation()
    test_classify()
    test_sse_parsing()
    test_reasoning_storage()
    test_migration_adds_reasoning()
    test_user_content()
    test_prompt_store()
    test_token_estimate()
    test_usage_parsing()
    test_session_prompt_storage()
    test_prompt_column_migration()
    test_markdown_rendering()
    test_unbalanced_backticks()
    test_sidebar_toggle()
    test_shortcuts()
    test_system_qt_theme_guard()
    test_version_matches_changelog()
    test_venv_discovery()
    test_config_defaults()
    test_search_tool_schema()
    test_search_results_sanitising()
    test_tool_call_accumulation()
    test_search_loop_cap()
    test_search_failure_paths()
    test_search_storage()
    test_date_note()
    test_search_html()
    print("\nall checks passed")