05 — ТЕСТЫ: Проверка каждого слоя

Стек состоит из 4 сервисов:
Ollama → Router → ChromaDB → WebUI

Если что-то сломалось — нужно знать ЧТО именно.
Без тестов: "ничего не работает, не знаю почему"
С тестами:  "Ollama ✅  Router ❌  — проблема в роутере"

Тесты запускаются:
- После установки (убедиться что всё ок)
- После перезагрузки сервера
- После обновления
- При любых проблемах

Каждый тест проверяет один конкретный факт:

УРОВЕНЬ 1: Сервис живой?          (HTTP 200)
УРОВЕНЬ 2: Сервис работает?       (правильный ответ)
УРОВЕНЬ 3: Логика правильная?     (coder для кода)
УРОВЕНЬ 4: Интеграция работает?   (Router → Ollama)

Тест = функция без аргументов,
       которая бросает Exception если что-то не так
       и ничего не возвращает если всё ок

# tests/test_stack.py
"""
Полное тестирование стека Local AI.

Запуск: python3 tests/test_stack.py
Или:    bash tests/run_tests.sh

Выход:
  0 — все тесты прошли
  1 — есть ошибки
"""
import requests
import sys
import time
import json

# Адреса сервисов
OLLAMA = "http://localhost:11434"
ROUTER = "http://localhost:8000"
WEBUI  = "http://localhost:3000"
CHROMA = "http://localhost:8001"

# Цвета для красивого вывода
GREEN  = "\033[92m"
RED    = "\033[91m"
YELLOW = "\033[93m"
RESET  = "\033[0m"

# Список результатов: (название, прошёл/нет)
results = []


def test(name: str, fn):
    """
    Запустить один тест и записать результат.

    name: описание что проверяем
    fn:   функция без аргументов, бросает Exception при ошибке
    """
    try:
        fn()
        print(f"{GREEN}✅{RESET} {name}")
        results.append((name, True))
    except AssertionError as e:
        print(f"{RED}❌{RESET} {name}: {e}")
        results.append((name, False))
    except requests.ConnectionError:
        print(f"{RED}❌{RESET} {name}: сервис недоступен")
        results.append((name, False))
    except requests.Timeout:
        print(f"{RED}❌{RESET} {name}: таймаут")
        results.append((name, False))
    except Exception as e:
        print(f"{RED}❌{RESET} {name}: {type(e).__name__}: {e}")
        results.append((name, False))


# ═══════════════════════════════════════════════════════════════
# ТЕСТЫ OLLAMA
# ═══════════════════════════════════════════════════════════════

def check_ollama_running():
    """
    Проверить что Ollama API отвечает.
    GET /api/tags должен вернуть 200.
    """
    r = requests.get(f"{OLLAMA}/api/tags", timeout=5)
    assert r.status_code == 200, f"Статус код: {r.status_code}"


def check_qwen_loaded():
    """
    Проверить что основная модель qwen2.5 скачана.
    Без неё ничего не работает.
    """
    r = requests.get(f"{OLLAMA}/api/tags", timeout=5)
    models = [m["name"] for m in r.json().get("models", [])]
    qwen_models = [m for m in models if "qwen2.5" in m]
    assert len(qwen_models) > 0, \
        f"qwen2.5 не найден. Есть: {models}"


def check_embed_model_loaded():
    """
    Проверить что модель embeddings загружена.
    Нужна для RAG и долгой памяти.
    """
    r = requests.get(f"{OLLAMA}/api/tags", timeout=5)
    models = [m["name"] for m in r.json().get("models", [])]
    embed_models = [m for m in models if "nomic" in m or "embed" in m]
    assert len(embed_models) > 0, \
        f"embed модель не найдена. Есть: {models}"


def check_ollama_generates():
    """
    Проверить что модель реально генерирует текст.
    Это самый важный тест — если он падает, ничего не работает.

    Используем простой запрос с коротким ожидаемым ответом.
    Таймаут 60 секунд — первая генерация медленнее (загрузка).
    """
    r = requests.post(f"{OLLAMA}/api/generate", json={
        "model": "qwen2.5:14b",
        "prompt": "Ответь одним словом: 2+2=",
        "stream": False,
        "options": {"num_predict": 10}  # максимум 10 токенов
    }, timeout=60)

    assert r.status_code == 200, f"Статус: {r.status_code}"
    response = r.json().get("response", "")
    assert len(response) > 0, "Пустой ответ от модели"


def check_embed_works():
    """
    Проверить что модель embeddings создаёт векторы.
    Вектор должен быть списком из 768 чисел.
    """
    r = requests.post(f"{OLLAMA}/api/embed", json={
        "model": "nomic-embed-text",
        "input": "тестовый текст"
    }, timeout=30)

    assert r.status_code == 200
    embeddings = r.json().get("embeddings", [])
    assert len(embeddings) > 0, "Нет embeddings в ответе"
    assert len(embeddings[0]) > 100, \
        f"Вектор слишком короткий: {len(embeddings[0])}"


# ═══════════════════════════════════════════════════════════════
# ТЕСТЫ ROUTER
# ═══════════════════════════════════════════════════════════════

def check_router_running():
    """Роутер отвечает на health check."""
    r = requests.get(f"{ROUTER}/health", timeout=5)
    assert r.status_code == 200
    data = r.json()
    assert data.get("status") in ["ok", "degraded"], \
        f"Неожиданный статус: {data}"


def check_router_picks_code_model():
    """
    Роутер должен выбирать code модель для запросов про код.

    Это критически важная логика — без неё всегда используется
    одна модель и нет преимущества специализации.
    """
    r = requests.post(f"{ROUTER}/chat", json={
        "prompt": "напиши python функцию сортировки"
    }, timeout=5)  # не ждём ответа модели — только проверяем выбор

    # Если таймаут на генерацию — это ок для этого теста
    # Нас интересует только какая модель выбрана
    # Поэтому используем отдельный эндпоинт

    # Простая проверка через логику pick_model напрямую
    # (без реального запроса к модели)
    r2 = requests.post(f"{ROUTER}/pick-model",
                       json={"prompt": "напиши python функцию"}, timeout=5)
    if r2.status_code == 200:
        model = r2.json().get("model", "")
        assert "coder" in model, \
            f"Ожидался coder, выбрана: {model}"


def check_router_picks_default():
    """
    Для обычных вопросов роутер должен использовать default модель.
    """
    r = requests.post(f"{ROUTER}/pick-model",
                      json={"prompt": "что такое фотосинтез?"}, timeout=5)
    if r.status_code == 200:
        model = r.json().get("model", "")
        assert "coder" not in model, \
            f"Для обычного вопроса выбрана coder модель: {model}"


def check_router_responds():
    """
    Роутер возвращает ответ на простой вопрос.
    Полный цикл: Router → Ollama → Router.
    """
    r = requests.post(f"{ROUTER}/chat", json={
        "prompt": "скажи только слово тест"
    }, timeout=90)  # 90 секунд — модель может долго грузиться

    assert r.status_code == 200, f"Статус: {r.status_code}"
    data = r.json()
    assert "response" in data, f"Нет поля response: {data}"
    assert len(data["response"]) > 0, "Пустой ответ"
    assert "model_used" in data, "Нет поля model_used"


# ═══════════════════════════════════════════════════════════════
# ТЕСТЫ CHROMADB
# ═══════════════════════════════════════════════════════════════

def check_chromadb_running():
    """ChromaDB отвечает на heartbeat."""
    r = requests.get(f"{CHROMA}/api/v1/heartbeat", timeout=5)
    assert r.status_code == 200


def check_chromadb_save_search():
    """
    Полный цикл памяти: сохранить → найти.

    1. Сохраняем тестовый факт
    2. Ищем похожее
    3. Проверяем что нашли
    """
    import uuid

    # 1. Создать тестовую коллекцию
    test_collection = f"test_{uuid.uuid4().hex[:8]}"
    requests.post(f"{CHROMA}/api/v1/collections",
                  json={"name": test_collection})

    # 2. Получить вектор через Ollama
    r = requests.post(f"{OLLAMA}/api/embed", json={
        "model": "nomic-embed-text",
        "input": "тестовый документ о Python программировании"
    }, timeout=30)
    vector = r.json()["embeddings"][0]

    # 3. Сохранить
    requests.post(f"{CHROMA}/api/v1/collections/{test_collection}/add",
                  json={
                      "ids": ["test-1"],
                      "embeddings": [vector],
                      "documents": ["тестовый документ о Python"]
                  })

    # 4. Поиск
    query_r = requests.post(f"{OLLAMA}/api/embed", json={
        "model": "nomic-embed-text",
        "input": "Python код"
    }, timeout=30)
    query_vector = query_r.json()["embeddings"][0]

    search_r = requests.post(
        f"{CHROMA}/api/v1/collections/{test_collection}/query",
        json={"query_embeddings": [query_vector], "n_results": 1})

    docs = search_r.json().get("documents", [[]])[0]
    assert len(docs) > 0, "Ничего не найдено в ChromaDB"
    assert "Python" in docs[0], f"Нашли не то: {docs[0]}"

    # 5. Очистить за собой
    requests.delete(f"{CHROMA}/api/v1/collections/{test_collection}")


# ═══════════════════════════════════════════════════════════════
# ТЕСТЫ OPEN WEBUI
# ═══════════════════════════════════════════════════════════════

def check_webui_running():
    """Open WebUI отдаёт HTML страницу."""
    r = requests.get(WEBUI, timeout=10)
    assert r.status_code == 200
    assert "html" in r.headers.get("content-type", "").lower(), \
        "Не HTML ответ"


# ═══════════════════════════════════════════════════════════════
# ЗАПУСК ВСЕХ ТЕСТОВ
# ═══════════════════════════════════════════════════════════════

def run_all():
    print(f"\n{'='*50}")
    print("ТЕСТ СТЕКА LOCAL AI")
    print(f"{'='*50}\n")

    print("─── Ollama ───────────────────────────────────")
    test("Ollama запущен",                   check_ollama_running)
    test("Qwen2.5 загружен",                 check_qwen_loaded)
    test("Embed модель загружена",            check_embed_model_loaded)
    test("Ollama генерирует текст",          check_ollama_generates)
    test("Embeddings работают",              check_embed_works)

    print("\n─── Router ───────────────────────────────────")
    test("Router запущен",                   check_router_running)
    test("Выбирает coder для python кода",   check_router_picks_code_model)
    test("Выбирает default для вопросов",    check_router_picks_default)
    test("Router отвечает на запрос",        check_router_responds)

    print("\n─── ChromaDB ─────────────────────────────────")
    test("ChromaDB запущен",                 check_chromadb_running)
    test("Сохранение и поиск работает",      check_chromadb_save_search)

    print("\n─── Open WebUI ───────────────────────────────")
    test("WebUI доступен",                   check_webui_running)

    # Итог
    passed = sum(1 for _, ok in results if ok)
    total  = len(results)
    failed_tests = [name for name, ok in results if not ok]

    print(f"\n{'='*50}")
    if passed == total:
        print(f"{GREEN}✅ Все {total}/{total} тестов прошли{RESET}")
        print("\nСтек работает корректно!")
        print(f"Интерфейс: http://localhost:3000")
    else:
        print(f"{RED}❌ {passed}/{total} тестов прошли{RESET}")
        print(f"\nПроблемные тесты:")
        for name in failed_tests:
            print(f"  - {name}")
        print("\nСм. 03_INSTALL.md → Решение проблем")

    print('='*50)

    return 0 if passed == total else 1


if __name__ == "__main__":
    sys.exit(run_all())

#!/bin/bash
# tests/run_tests.sh
# Запуск тестов с ожиданием старта сервисов

echo "=== ТЕСТ LOCAL AI СТЕКА ==="
echo ""

# Ждём пока сервисы поднимутся
echo "Ожидание запуска сервисов (до 60 секунд)..."

for i in $(seq 1 12); do
    if curl -s http://localhost:11434/api/tags > /dev/null 2>&1; then
        echo "✅ Ollama готов"
        break
    fi
    echo "  Попытка $i/12..."
    sleep 5
done

for i in $(seq 1 6); do
    if curl -s http://localhost:8000/health > /dev/null 2>&1; then
        echo "✅ Router готов"
        break
    fi
    sleep 5
done

echo ""

# Запустить тесты
python3 tests/test_stack.py
EXIT_CODE=$?

echo ""

# Показать логи если есть ошибки
if [ $EXIT_CODE -ne 0 ]; then
    echo "=== Логи сервисов ==="
    echo ""
    echo "--- Ollama ---"
    journalctl -u ollama -n 20 --no-pager

    echo ""
    echo "--- Router ---"
    docker logs ai-router --tail=20 2>/dev/null

    echo ""
    echo "--- ChromaDB ---"
    docker logs chromadb --tail=10 2>/dev/null
fi

exit $EXIT_CODE

# Ollama живой?
curl http://localhost:11434/api/tags
# Ожидание: {"models":[...]}

# Какие модели загружены?
curl http://localhost:11434/api/tags | python3 -c "
import json,sys
data = json.load(sys.stdin)
for m in data['models']:
    size_gb = m['size'] / 1024**3
    print(f\"{m['name']} ({size_gb:.1f}GB)\")
"

# Тест генерации напрямую
curl -s http://localhost:11434/api/generate \
  -d '{"model":"qwen2.5:14b","prompt":"1+1=","stream":false}' \
| python3 -c "import json,sys; print(json.load(sys.stdin)['response'])"

# Router здоров?
curl http://localhost:8000/health | python3 -m json.tool

# ChromaDB живой?
curl http://localhost:8001/api/v1/heartbeat

# Полный запрос через Router
curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt":"скажи привет"}' \
| python3 -c "
import json,sys
data = json.load(sys.stdin)
print('Модель:', data['model_used'])
print('Ответ:', data['response'][:100])
"

# Сколько RAM занимает Ollama
ps aux | grep ollama | awk '{print $6/1024 " MB RAM"}'

# Какие модели загружены в память прямо сейчас
curl http://localhost:11434/api/ps

# Температура CPU (если есть sensors)
sensors | grep -i "core\|temp"

# Нагрузка на CPU при генерации
htop  # смотреть в реальном времени

# Логи Ollama в реальном времени
journalctl -u ollama -f

# Логи Router
docker logs ai-router -f

# Все контейнеры
docker-compose ps

# Использование дискового пространства моделями
du -sh ~/.ollama/models/

Тесты покрывают:
✅ Ollama запущен и отвечает
✅ Нужные модели скачаны
✅ Генерация работает
✅ Embeddings работают
✅ Router запущен и выбирает правильные модели
✅ ChromaDB сохраняет и ищет
✅ WebUI доступен

Если все тесты прошли — стек работает корректно.
Можно открывать http://localhost:3000 и работать.