omnisvg-lite/tests/test_history.py

"""Юнит-тесты для history.py.

Покрывают:
- add(): вставка записи возвращает id, JSON-поля сериализуются
- list_recent(): сортировка DESC, лимит, новые сверху
- get(): по id, None для несуществующего
- persist: создать/закрыть/создать — данные сохранились на диске
- count(): корректный счёт
- Спецсимволы в prompt: юникод, кавычки, переносы строк, эмодзи
- JSON-поля: list[str] с не-ASCII корректно декодируется обратно
- WAL mode: PRAGMA journal_mode после открытия соединения
- list_recent(limit=...) ограничивает выборку

Каждый тест использует tmp_path (pytest fixture) для изолированной БД.

Запуск: `python -m pytest tests/test_history.py -v`
"""

from __future__ import annotations

import sqlite3
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import pytest  # noqa: E402

from history import History, Record  # noqa: E402


# ---------------------------------------------------------------------------
# Базовые CRUD
# ---------------------------------------------------------------------------


def _make_record(**overrides) -> Record:
    """Минимальный валидный Record + overrides."""
    base = dict(
        prompt="a fox",
        mode="icon",
        model="test-model",
        n_requested=2,
        n_returned=2,
        temperature=0.4,
        status="ok",
        error_reason=None,
        raw_outputs=["raw1", "raw2"],
        validated_outputs=["<svg/>", "<svg/>"],
        previews=["/p/1.png", "/p/2.png"],
        best_index=0,
    )
    base.update(overrides)
    return Record(**base)


def test_add_returns_increasing_ids(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        id1 = h.add(_make_record(prompt="a"))
        id2 = h.add(_make_record(prompt="b"))
        id3 = h.add(_make_record(prompt="c"))
    assert id1 == 1
    assert id2 == 2
    assert id3 == 3
    assert id1 < id2 < id3


def test_get_returns_record_by_id(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        rid = h.add(_make_record(prompt="hello"))
    with History(db) as h:
        rec = h.get(rid)
    assert rec is not None
    assert rec["id"] == rid
    assert rec["prompt"] == "hello"
    assert rec["mode"] == "icon"
    assert rec["status"] == "ok"
    assert rec["n_requested"] == 2
    assert rec["n_returned"] == 2
    assert rec["temperature"] == 0.4
    assert rec["raw_outputs"] == ["raw1", "raw2"]
    assert rec["validated_outputs"] == ["<svg/>", "<svg/>"]
    assert rec["previews"] == ["/p/1.png", "/p/2.png"]
    assert rec["best_index"] == 0


def test_get_returns_none_for_missing_id(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        assert h.get(99999) is None


def test_count_returns_total(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        assert h.count() == 0
        h.add(_make_record(prompt="a"))
        h.add(_make_record(prompt="b"))
        h.add(_make_record(prompt="c"))
        assert h.count() == 3


# ---------------------------------------------------------------------------
# list_recent
# ---------------------------------------------------------------------------


def test_list_recent_returns_newest_first(tmp_path: Path):
    """Новые записи — сверху, лимит работает."""
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        for i in range(5):
            h.add(_make_record(prompt=f"prompt-{i}"))
            time.sleep(0.005)  # гарантируем уникальный created_at
    with History(db) as h:
        rows = h.list_recent(limit=3)
    assert len(rows) == 3
    # Новейшие сверху → "prompt-4", "prompt-3", "prompt-2"
    assert [r["prompt"] for r in rows] == ["prompt-4", "prompt-3", "prompt-2"]


def test_list_recent_default_limit_is_50(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        for i in range(60):
            h.add(_make_record(prompt=f"p-{i}"))
    with History(db) as h:
        rows = h.list_recent()
    assert len(rows) == 50


def test_list_recent_handles_explicit_large_limit(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        for i in range(3):
            h.add(_make_record(prompt=f"p-{i}"))
    with History(db) as h:
        rows = h.list_recent(limit=1000)
    assert len(rows) == 3


# ---------------------------------------------------------------------------
# Persist между открытиями
# ---------------------------------------------------------------------------


def test_data_persists_across_open_close(tmp_path: Path):
    """Создать → закрыть → снова открыть → данные на месте."""
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        rid1 = h.add(_make_record(prompt="first"))
        rid2 = h.add(_make_record(prompt="second"))
    # БД закрыта; открываем заново
    assert db.exists()
    assert db.stat().st_size > 0
    with History(db) as h:
        assert h.count() == 2
        assert h.get(rid1)["prompt"] == "first"
        assert h.get(rid2)["prompt"] == "second"


def test_data_survives_full_reopen_with_wal_files(tmp_path: Path):
    """WAL-файлы могут остаться после закрытия — повторное открытие должно работать."""
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        h.add(_make_record(prompt="x"))
    # Проверяем, что нет orphan'ов: всё читается.
    with History(db) as h:
        rec = h.get(1)
        assert rec["prompt"] == "x"


# ---------------------------------------------------------------------------
# Спецсимволы в prompt
# ---------------------------------------------------------------------------


def test_prompt_with_unicode_cyrillic(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    cyrillic = "лиса в осеннем лесу, палитра оранжевая"
    with History(db) as h:
        rid = h.add(_make_record(prompt=cyrillic))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["prompt"] == cyrillic


def test_prompt_with_quotes_and_doublequotes(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    p = 'icon with "double" and \'single\' quotes'
    with History(db) as h:
        rid = h.add(_make_record(prompt=p))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["prompt"] == p


def test_prompt_with_newlines_and_tabs(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    p = "line 1\nline 2\n\tindented\n\nblank-line-above"
    with History(db) as h:
        rid = h.add(_make_record(prompt=p))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["prompt"] == p


def test_prompt_with_emoji(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    p = "fox 🦊 in forest 🌲🌲"
    with History(db) as h:
        rid = h.add(_make_record(prompt=p))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["prompt"] == p


def test_prompt_with_backslashes_and_sql_injection_attempt(tmp_path: Path):
    """Промпт с SQL-инъекцией в виде текста — должен храниться как plain text."""
    db = tmp_path / "h.sqlite"
    evil = "'; DROP TABLE generations; --"
    with History(db) as h:
        rid = h.add(_make_record(prompt=evil))
    with History(db) as h2:
        rec = h2.get(rid)
        assert rec["prompt"] == evil
        # Таблица жива
        assert h2.count() == 1


def test_prompt_with_very_long_string(tmp_path: Path):
    """Длинный промпт (10K символов) — должен сохраниться без потерь."""
    db = tmp_path / "h.sqlite"
    p = "x" * 10000
    with History(db) as h:
        rid = h.add(_make_record(prompt=p))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["prompt"] == p
    assert len(rec["prompt"]) == 10000


# ---------------------------------------------------------------------------
# JSON-поля с не-ASCII
# ---------------------------------------------------------------------------


def test_raw_outputs_with_unicode_preserved(tmp_path: Path):
    """list[str] в raw_outputs хранится как JSON, ensure_ascii=False."""
    db = tmp_path / "h.sqlite"
    raw = [
        '<svg viewBox="0 0 64 64"><text>лиса 🦊</text></svg>',
        '<svg viewBox="0 0 64 64"><text>simple</text></svg>',
    ]
    with History(db) as h:
        rid = h.add(_make_record(raw_outputs=raw))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["raw_outputs"] == raw


def test_validated_outputs_with_unicode_in_svg(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    svg = '<svg viewBox="0 0 512 512"><text>Привет</text></svg>'
    with History(db) as h:
        rid = h.add(_make_record(validated_outputs=[svg]))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["validated_outputs"] == [svg]


def test_previews_with_unicode_paths(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    paths = [
        "C:\\Users\\пользователь\\превью\\1_0.png",
        "D:\\AI\\Projects\\лиса\\2_1.png",
    ]
    with History(db) as h:
        rid = h.add(_make_record(previews=paths))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["previews"] == paths


# ---------------------------------------------------------------------------
# Контекст-менеджер: WAL и поведение при ошибках
# ---------------------------------------------------------------------------


def test_wal_mode_is_set(tmp_path: Path):
    """После открытия History journal_mode должен быть WAL."""
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        # Спросим напрямую через raw-conn.
        cur = h.conn.execute("PRAGMA journal_mode")
        mode = cur.fetchone()[0]
    assert mode.lower() == "wal"


def test_context_manager_closes_connection(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    h = History(db)
    with h as hist:
        hist.add(_make_record(prompt="x"))
    # После выхода conn=None
    assert h._conn is None


def test_context_manager_commit_path_runs_in_exit(tmp_path: Path):
    """Без exception — __exit__ коммитит оставшиеся незакоммиченные изменения.

    Замечание: `add()` сам вызывает commit() внутри, поэтому запись из add()
    сохранится в любом случае. Этот тест проверяет, что exit/close не
    портит уже закоммиченное и не падает на нормальном пути.
    """
    db = tmp_path / "h.sqlite"
    h = History(db)
    with h as hist:
        hist.add(_make_record(prompt="x"))
    # Контекст закрылся без exception
    assert h._conn is None
    with History(db) as hist2:
        assert hist2.count() == 1


def test_conn_outside_context_raises(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    h = History(db)
    with pytest.raises(RuntimeError, match="контекст-менеджера"):
        _ = h.conn


# ---------------------------------------------------------------------------
# Нишевые: пустые list'ы, edge-cases значений
# ---------------------------------------------------------------------------


def test_record_with_empty_lists(tmp_path: Path):
    """Record с пустыми raw/validated/previews — должен сохраниться."""
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        rid = h.add(_make_record(
            status="failed",
            error_reason="lm studio down",
            raw_outputs=[],
            validated_outputs=[],
            previews=[],
            best_index=None,
        ))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["raw_outputs"] == []
    assert rec["validated_outputs"] == []
    assert rec["previews"] == []
    assert rec["best_index"] is None
    assert rec["error_reason"] == "lm studio down"


def test_default_created_at_set_by_add(tmp_path: Path):
    """Если created_at=None в Record — add() заполняет time.time()."""
    db = tmp_path / "h.sqlite"
    before = time.time()
    with History(db) as h:
        rec_in = _make_record(prompt="x")
        assert rec_in.created_at is None
        rid = h.add(rec_in)
    after = time.time()
    with History(db) as h:
        rec_out = h.get(rid)
    assert before - 1 <= rec_out["created_at"] <= after + 1


def test_explicit_created_at_preserved(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    fixed_ts = 1700000000.0
    with History(db) as h:
        rid = h.add(_make_record(prompt="x", created_at=fixed_ts))
    with History(db) as h:
        rec = h.get(rid)
    assert rec["created_at"] == fixed_ts


# ---------------------------------------------------------------------------
# Несколько записей: list_recent с лимитом < N
# ---------------------------------------------------------------------------


def test_list_recent_limit_smaller_than_total(tmp_path: Path):
    db = tmp_path / "h.sqlite"
    with History(db) as h:
        for i in range(10):
            h.add(_make_record(prompt=f"p-{i}"))
            time.sleep(0.003)
    with History(db) as h:
        rows = h.list_recent(limit=2)
    assert len(rows) == 2
    # Самые свежие
    assert rows[0]["prompt"] == "p-9"
    assert rows[1]["prompt"] == "p-8"