"""Регрессионные тесты на callback-логику app.py.
Verifier feedback (attempt 1): "back crashes on input validation paths because
`gr.Warning` was changed from a class to a function in Gradio 5.x and the
producer didn't migrate. Happy path works. Unit tests don't cover this path.
Manual first-click on bad input would surface a TypeError."
Эти тесты ловят именно эту ошибку. Они НЕ дёргают Gradio UI — только
вызывают `app.on_generate` напрямую и проверяют, что:
1) Нет TypeError (т.е. внутри нет `raise gr.Warning/Error`).
2) Возвращается правильное число плейсхолдеров.
3) `app.on_generate` не пытается ходить в LM Studio, если входные данные
отклонены на pre-check.
Запуск: `python -m pytest tests/test_app.py -v`
"""
from __future__ import annotations
import sys
import time
from pathlib import Path
from typing import Any
from unittest.mock import patch
ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import pytest # noqa: E402
# ---------------------------------------------------------------------------
# Сторож: gr.Warning / gr.Error в Gradio 5.x — это функции, а не классы.
# Если кто-то когда-то обновит gradio и это поведение изменится — тест
# напомнит, что нужно пересмотреть on_generate.
# ---------------------------------------------------------------------------
def test_gradio_warning_is_function_not_class():
import inspect
import gradio as gr
# Контрактное свойство Gradio 5.x, на которое опирается on_generate:
# `gr.Warning(...)` — это ФУНКЦИЯ (а не класс исключения), и её нужно
# ВЫЗЫВАТЬ. Если в новой версии Gradio это поведение изменится, тест
# упадёт, и on_generate нужно будет пересмотреть.
assert inspect.isclass(gr.Warning) is False, (
"gr.Warning стал классом в этой версии Gradio — пересмотрите on_generate"
)
assert callable(gr.Warning)
# ---------------------------------------------------------------------------
# Хелпер: дёрнуть on_generate с разными входами и поймать TypeError.
# ---------------------------------------------------------------------------
def _call_on_generate(**overrides: Any) -> Any:
"""Вызывает app.on_generate с минимальным валидным набором + overrides.
Возвращает то, что вернул callback. Если внутри есть `raise gr.Warning`,
получим TypeError ещё ДО того, как вернётся значение.
"""
from app import on_generate
defaults: dict[str, Any] = dict(
prompt="filled magnifying glass", # валидный
mode="icon",
n_candidates=2,
temperature=0.4,
image=None,
palette="",
model="test-model",
base_url="http://127.0.0.1:1234/v1",
api_key="lm-studio",
)
defaults.update(overrides)
# Патчим chat() так, чтобы on_generate не уходил в сеть и не упал уже
# ВНЕ pre-check. Если pre-check пропустил и chat() зовётся — мы увидим
# ValueError от mock'а, что нас устраивает (это другая ветка).
with patch("app.chat") as mock_chat:
mock_chat.side_effect = RuntimeError("chat should not be called from this test")
return on_generate(**defaults)
def test_on_generate_does_not_raise_on_empty_prompt():
"""Критический регрессионный кейс: пустой промпт → gr.Warning (НЕ raise)."""
try:
result = _call_on_generate(prompt="")
except TypeError as exc:
pytest.fail(
"on_generate упал с TypeError на пустом промпте — "
"вероятно, кто-то вернул `raise gr.Warning(...)`: "
f"{exc}"
)
# Должен вернуть 7 плейсхолдеров для outputs.
assert isinstance(result, tuple)
assert len(result) == 7, f"ожидался кортеж из 7 элементов, получено {len(result)}"
def test_on_generate_does_not_raise_on_too_long_prompt():
try:
_call_on_generate(prompt="x" * 1001)
except TypeError as exc:
pytest.fail(
f"on_generate упал с TypeError на длинном промпте: {exc}"
)
def test_on_generate_does_not_raise_on_bad_n_candidates():
try:
result = _call_on_generate(n_candidates=0)
except TypeError as exc:
pytest.fail(f"on_generate упал с TypeError на n=0: {exc}")
assert isinstance(result, tuple) and len(result) == 7
# n=99 — тоже вне диапазона
try:
_call_on_generate(n_candidates=99)
except TypeError as exc:
pytest.fail(f"on_generate упал с TypeError на n=99: {exc}")
def test_on_generate_does_not_raise_on_bad_mode():
try:
result = _call_on_generate(mode="portrait")
except TypeError as exc:
pytest.fail(f"on_generate упал с TypeError на неизвестном mode: {exc}")
assert isinstance(result, tuple) and len(result) == 7
def test_on_generate_does_not_call_chat_on_precheck_fail():
"""Если pre-check упал, chat() НЕ должен вызываться вообще."""
from app import on_generate
with patch("app.chat") as mock_chat:
on_generate(
prompt="", # упадёт на pre-check
mode="icon",
n_candidates=2,
temperature=0.4,
image=None,
palette="",
model="x",
base_url="http://127.0.0.1:1234/v1",
api_key="lm-studio",
)
assert mock_chat.call_count == 0, (
"chat() был вызван, хотя pre-check должен был остановить поток"
)
def test_on_generate_does_not_call_chat_on_bad_mode():
from app import on_generate
with patch("app.chat") as mock_chat:
on_generate(
prompt="valid",
mode="junk",
n_candidates=2,
temperature=0.4,
image=None,
palette="",
model="x",
base_url="http://127.0.0.1:1234/v1",
api_key="lm-studio",
)
assert mock_chat.call_count == 0
# ---------------------------------------------------------------------------
# Положительный smoke: на корректном входе on_generate НЕ возвращает пустоту
# (хотя в этом юнит-тесте chat() замокан → идём по ветке ошибки сборки
# промпта, а не успеха; это нормально, главное — нет TypeError).
# ---------------------------------------------------------------------------
def test_on_generate_with_valid_input_does_not_typeerror():
"""Даже когда chat() падает (замокан), pre-check не должен давать TypeError."""
from app import on_generate
with patch("app.chat") as mock_chat:
mock_chat.return_value = type("R", (), {
"raw_texts": [''],
"elapsed_s": 0.1,
"model": "test",
"usage": None,
"finish_reasons": ["stop"],
})()
try:
result = on_generate(
prompt="filled magnifying glass",
mode="icon",
n_candidates=1,
temperature=0.4,
image=None,
palette="",
model="test",
base_url="http://127.0.0.1:1234/v1",
api_key="lm-studio",
)
except TypeError as exc:
pytest.fail(f"on_generate упал с TypeError на валидном входе: {exc}")
# На валидном входе возвращается кортеж из 7 элементов.
assert isinstance(result, tuple)
assert len(result) == 7
# ---------------------------------------------------------------------------
# Smoke-тест: импорт app и build_ui() возвращает gr.Blocks.
# Не лезем в сеть, не запускаем UI.
# ---------------------------------------------------------------------------
def test_app_module_imports():
"""app.py импортируется без ошибок (все зависимости в порядке)."""
import app # noqa: F401
assert hasattr(app, "on_generate")
assert hasattr(app, "on_history_select")
assert hasattr(app, "build_ui")
assert hasattr(app, "main")
def test_build_ui_returns_gradio_blocks():
"""build_ui() возвращает gr.Blocks (smoke-тест сборки UI)."""
from app import build_ui
demo = build_ui()
# Проверяем, что это действительно gr.Blocks, а не None или что-то другое.
import gradio as gr
assert isinstance(demo, gr.Blocks), f"ожидался gr.Blocks, получено {type(demo).__name__}"
def test_on_mode_change_returns_icon_default_n():
from app import on_mode_change
update = on_mode_change("icon")
# gr.update — это dict-like объект, у него есть .value
assert update["value"] == 1
def test_on_mode_change_returns_illustration_default_n():
from app import on_mode_change
update = on_mode_change("illustration")
assert update["value"] == 1
# ---------------------------------------------------------------------------
# Live-режим (on_generate_live)
#
# Пять обязательных тестов из ТЗ live-ui:
# 1. yields ≥ 2 промежуточных превью при 5 дельтах
# 2. финальный yield содержит status "Сгенерировано" и валидный SVG
# 3. throttle режет быстрые дельты (10 за 50мс → ≤ 4-5 yield'ов)
# 4. после end event в SQLite появляется запись (status=ok/partial)
# 5. при LMStudioUnavailable вызывается gr.Error
#
# Плюс sanity-тест: on_generate_live — генератор-функция.
# ---------------------------------------------------------------------------
# Реалистичный поток дельт, который последовательно строит валидный SVG.
# Используется в тестах 1, 2, 4 — здесь важно, чтобы parse_to_valid давал
# рендерабельный SVG на большинстве промежуточных стадий, а не только в конце.
_VALID_SVG_DELTAS = [
"",
]
def _make_stream_mock(deltas, *, sleep_s: float = 0.20):
"""Возвращает мок-функцию stream_chat, отдающую дельты + 'end'.
Args:
deltas: список строк-дельта.
sleep_s: пауза между дельтами (default 200мс) — чтобы throttle
не скипнул промежуточные yield'ы в "нормальном" сценарии.
"""
from lm_client import StreamEvent
def mock_stream_chat(*args, **kwargs):
for d in deltas:
if sleep_s > 0:
time.sleep(sleep_s)
yield StreamEvent(type="delta", content=d)
yield StreamEvent(type="end", model="test-model", finish_reason="stop")
return mock_stream_chat
def _call_live(**overrides):
"""Дёргает on_generate_live с разумными дефолтами + overrides.
Возвращает список yields (т.е. материализованный генератор).
"""
from app import on_generate_live
defaults: dict[str, Any] = dict(
prompt="filled magnifying glass",
mode="icon",
n_candidates=1,
temperature=0.4,
image=None,
palette="",
model="test-model",
base_url="http://127.0.0.1:1234/v1",
api_key="lm-studio",
use_live=True,
)
defaults.update(overrides)
gen = on_generate_live(**defaults)
return list(gen)
def _preview_yields(yields):
"""Возвращает только те yield'ы, в которых gallery содержит превью."""
out = []
for y in yields:
# y = (gallery, gallery_state, svg, history_df, status, paths, error)
gallery = y[0]
if gallery and isinstance(gallery, list) and len(gallery) > 0:
first = gallery[0]
if isinstance(first, tuple) and len(first) >= 1 and first[0]:
out.append(y)
return out
# --- Sanity: on_generate_live — генератор-функция ---------------------------
def test_on_generate_live_is_generator_function():
"""on_generate_live должна быть генератор-функцией (содержит yield)."""
import inspect
from app import on_generate_live
assert inspect.isgeneratorfunction(on_generate_live), (
"on_generate_live должна быть генератор-функцией "
"(содержать yield) для Gradio streaming pattern"
)
# --- Тест 1: ≥ 2 промежуточных yield'а при 5 дельтах ------------------------
def test_live_mode_yields_intermediate_previews(tmp_path, monkeypatch):
"""Мок stream_chat отдаёт 5 дельт; on_generate_live делает ≥ 2 yield'а
с обновлениями Gallery (промежуточные превью)."""
monkeypatch.setenv("OMNISVG_DB_PATH", str(tmp_path / "history.sqlite"))
monkeypatch.setenv("OMNISVG_PREVIEW_DIR", str(tmp_path / "previews"))
mock = _make_stream_mock(_VALID_SVG_DELTAS, sleep_s=0.20)
with patch("app.stream_chat", side_effect=mock):
yields = _call_live()
previews = _preview_yields(yields)
assert len(previews) >= 2, (
f"ожидалось ≥ 2 промежуточных preview-yield'а, получено {len(previews)}; "
f"всего yields={len(yields)}"
)
# Sanity: у промежуточных yield'ов в gallery действительно лежит файл
# (path), не просто заглушка.
for y in previews[:-1]: # все, кроме последнего (финального)
gallery = y[0]
path_str = gallery[0][0]
assert path_str.endswith(".png"), f"ожидался .png путь, получено {path_str!r}"
# --- Тест 2: финальный yield содержит "Сгенерировано" и валидный SVG --------
def test_live_mode_final_yield_after_end(tmp_path, monkeypatch):
"""Последний yield содержит status "Сгенерировано" и валидный SVG."""
monkeypatch.setenv("OMNISVG_DB_PATH", str(tmp_path / "history.sqlite"))
monkeypatch.setenv("OMNISVG_PREVIEW_DIR", str(tmp_path / "previews"))
mock = _make_stream_mock(_VALID_SVG_DELTAS, sleep_s=0.20)
with patch("app.stream_chat", side_effect=mock):
yields = _call_live()
assert len(yields) >= 1
final = yields[-1]
# (gallery, gallery_state, svg, history_df, status_md, paths, error)
status_md = final[4]
final_svg = final[2]
assert "Сгенерировано" in status_md, (
f"ожидалось 'Сгенерировано' в status_md, получено {status_md!r}"
)
# Валидный SVG: парсится lxml'ом, начинается с .
assert final_svg, "финальный SVG не должен быть пустым"
assert "" in final_svg
from lxml import etree
etree.fromstring(final_svg.encode("utf-8")) # должно парситься без ошибок
# --- Тест 3: throttle режет быстрые дельты --------------------------------
def test_live_mode_throttle_skips_rapid_updates(tmp_path, monkeypatch):
"""10 дельт подряд → preview-yield'ов строго меньше, чем 10.
Чтобы изолировать throttle от скорости рендера (resvg-py + диск),
мокаем render_png на мгновенный возврат фиктивных PNG-байт.
Без throttle мы получили бы 10 preview-yield'ов; с throttle=0.15s
на быстром стриме (10 дельт за <50мс) — максимум 1-2 preview-yield'а.
"""
monkeypatch.setenv("OMNISVG_DB_PATH", str(tmp_path / "history.sqlite"))
monkeypatch.setenv("OMNISVG_PREVIEW_DIR", str(tmp_path / "previews"))
# Мок render_png: возвращает фейковые PNG-байты мгновенно, чтобы
# тест измерял ТОЛЬКО throttle, а не скорость resvg/диска.
fake_png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32 # фейк-PNG заголовок
with patch("app.render_png", return_value=fake_png):
fast_deltas = [
"",
]
mock = _make_stream_mock(fast_deltas, sleep_s=0.0)
started = time.monotonic()
with patch("app.stream_chat", side_effect=mock):
yields = _call_live()
elapsed_ms = (time.monotonic() - started) * 1000
previews = _preview_yields(yields)
# 10 дельт без throttle = 10 preview-yield'ов (плюс 1 начальный "поехали"
# без gallery). С throttle=0.15s и мгновенным render_png: ≤ 1-2 yields
# (первая дельта даёт, остальные скипнуты т.к. < 150мс).
# Ставим жёсткий потолок 5, чтобы тест не флакал.
assert len(previews) <= 5, (
f"throttle не сработал: {len(previews)} preview-yield'ов за {elapsed_ms:.0f}мс; "
f"предел 5"
)
# Sanity: если throttle был ВООБЩЕ выключен, было бы ~10. Проверяем,
# что превью-yield'ов сильно меньше количества дельт (= 10).
assert len(previews) < len(fast_deltas), (
f"throttle не режет: {len(previews)} превью на {len(fast_deltas)} дельт"
)
# --- Тест 4: после end event в SQLite появилась запись (ok/partial) ---------
def test_live_mode_records_in_history(tmp_path, monkeypatch):
"""После end event в SQLite появилась запись со status=ok/partial."""
monkeypatch.setenv("OMNISVG_DB_PATH", str(tmp_path / "history.sqlite"))
monkeypatch.setenv("OMNISVG_PREVIEW_DIR", str(tmp_path / "previews"))
mock = _make_stream_mock(_VALID_SVG_DELTAS, sleep_s=0.20)
with patch("app.stream_chat", side_effect=mock):
yields = _call_live()
# Генерируем запись. Проверяем, что в SQLite есть новая строка.
from history import History
with History() as h:
records = h.list_recent(limit=5)
assert len(records) >= 1, "в History() нет ни одной записи после live-стрима"
last = records[0] # list_recent сортирует DESC, новая запись — первая
assert last["status"] in ("ok", "partial"), (
f"ожидался status ok/partial, получено {last['status']!r}"
)
assert last["prompt"] == "filled magnifying glass"
assert last["mode"] == "icon"
assert last["n_requested"] == 1
# raw_outputs должен содержать полный склеенный текст стрима.
assert last["raw_outputs"], "raw_outputs пуст"
assert "".join(last["raw_outputs"]).startswith("