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 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 "gemma-4-12B-it" in presets
    assert presets["gemma-4-12B-it"].vision is True
    # Qwen3.5-9B has its mmproj line commented out with '#'.
    if "Qwen3.5-9B" in presets:
        assert presets["Qwen3.5-9B"].vision is False
    print("ok  real presets classification")


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
    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_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_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")


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_markdown_rendering()
    test_system_qt_theme_guard()
    test_version_matches_changelog()
    test_venv_discovery()
    test_config_defaults()
    print("\nall checks passed")