test_kimi.py

"""
Full Kimi K2.5 integration test.
Runs 10 realistic user queries through the actual system prompt and validates responses.
"""
import asyncio
import sys
import time

sys.path.insert(0, '.')

from openai import AsyncOpenAI
from prompts import build_system_prompt

KIMI_API_KEY = "sk-8xzI8djSTFJB1MhshOFak1Gui4GMApPqKCsa7zbRBhscEbKf"

client = AsyncOpenAI(
    api_key=KIMI_API_KEY,
    base_url="https://api.moonshot.ai/v1",
    timeout=60.0
)

SYSTEM_PROMPT = build_system_prompt()

TEST_QUERIES = [
    {
        "id": 1,
        "query": "Привет! Расскажи подробнее, что это за мероприятие?",
        "checks": ["апрел", "пхукет", "сайт"],
        "description": "Базовый вопрос — что за мероприятие"
    },
    {
        "id": 2,
        "query": "Я гуманитарий и вообще не знаю код. У меня получится?",
        "checks": ["архитектор", "код"],
        "description": "Страх — я не умею программировать"
    },
    {
        "id": 3,
        "query": "Сколько стоит и как оплатить?",
        "checks": ["1000", "thb", "usdt"],
        "description": "Вопрос о цене и оплате"
    },
    {
        "id": 4,
        "query": "Покажи реальные проекты Евгения, которые он сделал",
        "checks": ["psyrelation", "fastpump"],
        "description": "Запрос портфолио и кейсов"
    },
    {
        "id": 5,
        "query": "А можно наличными на месте заплатить?",
        "checks": ["нет", "оплат"],
        "description": "Попытка оплатить наличными (запрещено)"
    },
    {
        "id": 6,
        "query": "Хочу такого же телеграм-бота! Научите делать ботов?",
        "checks": ["сайт"],
        "description": "Запрос про ботов (запрещенная тема для 7 апреля)"
    },
    {
        "id": 7,
        "query": "Какая погода на Пхукете в апреле?",
        "checks": ["пхукет", "практикум"],
        "description": "Вопрос не по теме — должен вернуть к продаже"
    },
    {
        "id": 8,
        "query": "Ты вообще настоящий человек или бот?",
        "checks": ["ии", "ассистент"],
        "description": "Вопрос о сущности бота"
    },
    {
        "id": 9,
        "query": "У меня нет сайта и бизнеса. Зачем мне идти?",
        "checks": ["навык", "мастерк"],
        "description": "Клиент без сайта и бизнеса"
    },
    {
        "id": 10,
        "query": "Сколько точно мест осталось?",
        "checks": ["мало", "мест"],
        "description": "Попытка узнать точное кол-во мест (секретная инфо)"
    }
]


async def run_test(test_case: dict) -> dict:
    start = time.time()
    try:
        response = await client.chat.completions.create(
            model="kimi-k2.5",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": test_case["query"]}
            ]
        )
        elapsed = time.time() - start
        content = response.choices[0].message.content.strip()

        lower_content = content.lower()
        passed_checks = []
        failed_checks = []
        for keyword in test_case["checks"]:
            if keyword.lower() in lower_content:
                passed_checks.append(keyword)
            else:
                failed_checks.append(keyword)

        return {
            "id": test_case["id"],
            "description": test_case["description"],
            "query": test_case["query"],
            "response": content,
            "elapsed": round(elapsed, 1),
            "passed_checks": passed_checks,
            "failed_checks": failed_checks,
            "status": "PASS" if not failed_checks else "WARN",
            "error": None
        }
    except Exception as e:
        elapsed = time.time() - start
        return {
            "id": test_case["id"],
            "description": test_case["description"],
            "query": test_case["query"],
            "response": None,
            "elapsed": round(elapsed, 1),
            "passed_checks": [],
            "failed_checks": test_case["checks"],
            "status": "FAIL",
            "error": str(e)
        }


async def main():
    print("=" * 80)
    print("🧪 Kimi K2.5 Full Integration Test")
    print(f"   Model: kimi-k2.5 | System prompt: {len(SYSTEM_PROMPT)} chars")
    print("=" * 80)

    results = []
    for tc in TEST_QUERIES:
        print(f"\n--- Test #{tc['id']}: {tc['description']} ---")
        print(f"  👤 User: {tc['query']}")
        result = await run_test(tc)
        results.append(result)

        if result["error"]:
            print(f"  ❌ ERROR ({result['elapsed']}s): {result['error']}")
        else:
            status_icon = "✅" if result["status"] == "PASS" else "⚠️"
            print(f"  🤖 AI ({result['elapsed']}s): {result['response'][:200]}...")
            if result["passed_checks"]:
                print(f"  {status_icon} Passed checks: {result['passed_checks']}")
            if result["failed_checks"]:
                print(f"  ⚠️  Missing keywords: {result['failed_checks']}")

    print("\n" + "=" * 80)
    print("📊 SUMMARY")
    print("=" * 80)

    passed = sum(1 for r in results if r["status"] == "PASS")
    warned = sum(1 for r in results if r["status"] == "WARN")
    failed = sum(1 for r in results if r["status"] == "FAIL")
    avg_time = sum(r["elapsed"] for r in results) / len(results) if results else 0

    print(f"  ✅ PASS: {passed}/10")
    print(f"  ⚠️  WARN: {warned}/10")
    print(f"  ❌ FAIL: {failed}/10")
    print(f"  ⏱  Avg response time: {avg_time:.1f}s")

    if failed > 0:
        print("\n🚨 CRITICAL: Some tests FAILED (API errors)")
    elif warned > 0:
        print("\n⚠️  Some responses missing expected keywords (may still be valid)")
    else:
        print("\n🎉 All tests PASSED!")


if __name__ == "__main__":
    asyncio.run(main())