Beta/test_bot.py

#!/usr/bin/env python3
"""Tests unitaires pour Superviseur Bot."""

import sys
import os
import json
import asyncio
import tempfile
import shutil
import logging

# Add the project root to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('TestBot')


# ==================== Tests d'import ====================

def test_imports():
    """Test que tous les modules principaux peuvent être importés."""
    from core.bot import Bot
    from core.llm import LLMManager
    from core.action_executor import ActionExecutor
    from brain.memory import (
        add_interaction, build_context_for_model, flush_all,
        delete_user_memory, set_llm_manager, _extractive_summary
    )
    from brain.moderation import init_db, log_infraction
    from dotenv import load_dotenv
    import discord
    logger.info("✓ Tous les imports réussis")


def test_llm_manager_init():
    """Test l'initialisation du LLMManager."""
    from core.llm import LLMManager
    llm = LLMManager(model_name="test-model")
    assert llm.model_name == "test-model"
    assert llm.temperature == 0.5
    assert "localhost" in llm.base_url
    logger.info("✓ LLMManager initialisé correctement")


def test_bot_init():
    """Test l'initialisation du bot Beta."""
    from core.bot import Bot
    from core.llm import LLMManager
    import discord
    
    llm = LLMManager(model_name="test-model")
    intents = discord.Intents.default()
    intents.message_content = True
    
    bot = Bot(
        llama_model=llm,
        command_prefix="!",
        guild_id=123456789,
        system_prompt="Test prompt",
        intents=intents
    )
    assert bot.guild_id == 123456789
    assert bot.system_name == "Bêta"
    logger.info("✓ Bot Superviseur initialisé correctement")


# ==================== Tests LLM (JSON Extraction) ====================

def test_extract_json_simple_object():
    """Test extraction JSON simple."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    text = 'Voici mon analyse : {"thought": "test", "action": "NONE", "response": "Salut !"}'
    result = llm.extract_json_actions(text)
    assert result is not None
    data = json.loads(result)
    assert data["action"] == "NONE"
    assert data["response"] == "Salut !"
    logger.info("✓ Extraction JSON simple OK")


def test_extract_json_array():
    """Test extraction JSON tableau."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    text = 'Actions : [{"action": "CREATE_CHANNEL", "channel_name": "test"}, {"action": "CREATE_ROLE", "role_name": "mod"}]'
    result = llm.extract_json_actions(text)
    assert result is not None
    data = json.loads(result)
    assert isinstance(data, list)
    assert len(data) == 2
    assert data[0]["action"] == "CREATE_CHANNEL"
    logger.info("✓ Extraction JSON tableau OK")


def test_extract_json_with_comments():
    """Test extraction JSON avec commentaires GPT-OSS."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    text = '''Voici la réponse :
// Ceci est un commentaire
{"action": "KICK", "user_id": "123", "reason": "spam"}
// Autre commentaire'''
    result = llm.extract_json_actions(text)
    assert result is not None
    data = json.loads(result)
    assert data["action"] == "KICK"
    logger.info("✓ Extraction JSON avec commentaires OK")


def test_extract_json_none():
    """Test extraction quand pas de JSON."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    text = "Salut, comment ça va ? Pas de JSON ici."
    result = llm.extract_json_actions(text)
    assert result is None
    logger.info("✓ Absence de JSON détectée correctement")


def test_extract_json_empty():
    """Test extraction avec texte vide."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    assert llm.extract_json_actions("") is None
    assert llm.extract_json_actions(None) is None
    logger.info("✓ Texte vide géré correctement")


def test_extract_json_nested_braces():
    """Test extraction avec accolades imbriquées."""
    from core.llm import LLMManager
    llm = LLMManager()
    
    text = 'Analyse : {"action": "SEND_MESSAGE", "channel_id": "123", "response": "Contenu avec {des} accolades"}'
    result = llm.extract_json_actions(text)
    assert result is not None
    data = json.loads(result)
    assert data["action"] == "SEND_MESSAGE"
    logger.info("✓ Extraction JSON avec accolades imbriquées OK")


# ==================== Tests LLM (Payload) ====================

def test_build_payload():
    """Test la construction de payload Ollama."""
    from core.llm import LLMManager
    llm = LLMManager(model_name="test-model", temperature=0.3)
    
    payload = llm.build_payload(
        prompt="Test prompt",
        system_prompt="Test system"
    )
    
    assert payload["model"] == "test-model"
    assert payload["prompt"] == "Test prompt"
    assert payload["system"] == "Test system"
    assert payload["options"]["temperature"] == 0.3
    assert payload["options"]["num_ctx"] == 16384
    logger.info("✓ Payload Ollama construit correctement")


# ==================== Tests Mémoire ====================

def _run_async(coro):
    """Helper pour exécuter du code async dans les tests sync."""
    return asyncio.get_event_loop().run_until_complete(coro)


def test_memory_save_load():
    """Test sauvegarde et chargement mémoire."""
    import brain.memoire as mem
    
    test_dir = tempfile.mkdtemp()
    original_dir = mem.MEMORY_DIR
    mem.MEMORY_DIR = test_dir
    mem.memory_cache.clear()
    mem._dirty.clear()
    
    try:
        uid = "test_user_123"
        data = {"resume_global": "Test résumé", "historique_recent": []}
        _run_async(mem.save_user_memory_async(uid, data, persist_immediately=True))
        loaded = _run_async(mem.load_user_memory_async(uid))
        assert loaded == data, f"Load/save failed: {loaded} != {data}"
        logger.info("✓ Sauvegarde/chargement mémoire OK")
    finally:
        mem.MEMORY_DIR = original_dir
        mem.memory_cache.clear()
        shutil.rmtree(test_dir)


def test_memory_add_interaction():
    """Test ajout d'interaction."""
    import brain.memoire as mem
    
    test_dir = tempfile.mkdtemp()
    original_dir = mem.MEMORY_DIR
    mem.MEMORY_DIR = test_dir
    mem.memory_cache.clear()
    mem._dirty.clear()
    
    try:
        uid = "test_user_456"
        _run_async(mem.add_interaction(uid, 111, "Hello", "Hi there!"))
        loaded = _run_async(mem.load_user_memory_async(uid))
        assert len(loaded["historique_recent"]) == 1
        assert loaded["historique_recent"][0]["user_message"] == "Hello"
        assert loaded["historique_recent"][0]["bot_message"] == "Hi there!"
        logger.info("✓ Ajout d'interaction OK")
    finally:
        mem.MEMORY_DIR = original_dir
        mem.memory_cache.clear()
        shutil.rmtree(test_dir)


def test_memory_build_context():
    """Test construction du contexte pour le modèle."""
    import brain.memoire as mem
    
    test_dir = tempfile.mkdtemp()
    original_dir = mem.MEMORY_DIR
    mem.MEMORY_DIR = test_dir
    mem.memory_cache.clear()
    mem._dirty.clear()
    
    try:
        uid = "test_user_789"
        _run_async(mem.add_interaction(uid, 111, "Question?", "Réponse!"))
        ctx = _run_async(mem.build_context_for_model(uid))
        assert "Question?" in ctx
        assert "Réponse!" in ctx
        logger.info("✓ Construction contexte OK")
    finally:
        mem.MEMORY_DIR = original_dir
        mem.memory_cache.clear()
        shutil.rmtree(test_dir)


def test_memory_delete():
    """Test suppression mémoire avec guild_id."""
    import brain.memoire as mem
    
    test_dir = tempfile.mkdtemp()
    original_dir = mem.MEMORY_DIR
    mem.MEMORY_DIR = test_dir
    mem.memory_cache.clear()
    mem._dirty.clear()
    
    try:
        uid = "test_user_del"
        _run_async(mem.add_interaction(uid, 111, "Test", "Response", guild_id="TestGuild"))
        
        # Forcer la sauvegarde immédiate (add_interaction schedule en background)
        _run_async(asyncio.sleep(0.5))
        
        # Vérifier que le fichier existe
        path = mem._user_filepath(uid, "TestGuild")
        assert os.path.exists(path), f"Fichier non trouvé: {path}"
        
        # Supprimer
        mem.delete_user_memory(uid, "TestGuild")
        
        # Vérifier suppression (recharger crée un fichier vide)
        mem.memory_cache.clear()
        loaded = _run_async(mem.load_user_memory_async(uid, "TestGuild"))
        assert loaded["historique_recent"] == []
        logger.info("✓ Suppression mémoire avec guild_id OK")
    finally:
        mem.MEMORY_DIR = original_dir
        mem.memory_cache.clear()
        shutil.rmtree(test_dir)


def test_memory_extractive_summary():
    """Test le résumé extractif (fallback)."""
    from brain.memoire import _extractive_summary
    
    messages = [
        {"user_message": "Salut, comment configurer le bot ?", "bot_message": "Tu peux utiliser la commande !config."},
        {"user_message": "Merci beaucoup !", "bot_message": "De rien, n'hésite pas."},
    ]
    summary = _extractive_summary(messages)
    assert len(summary) > 0
    assert "Mots-clés:" in summary
    logger.info("✓ Résumé extractif OK")


def test_memory_flush_all():
    """Test flush de toutes les mémoires dirty."""
    import brain.memoire as mem
    
    test_dir = tempfile.mkdtemp()
    original_dir = mem.MEMORY_DIR
    mem.MEMORY_DIR = test_dir
    mem.memory_cache.clear()
    mem._dirty.clear()
    
    try:
        # Créer quelques entrées
        _run_async(mem.add_interaction("user_a", 111, "Test A", "Response A"))
        _run_async(mem.add_interaction("user_b", 222, "Test B", "Response B"))
        
        # Marquer dirty
        assert any(mem._dirty.values())
        
        # Flush
        _run_async(mem.flush_all())
        
        # Vérifier que tout est clean
        assert not any(mem._dirty.values())
        logger.info("✓ Flush mémoire OK")
    finally:
        mem.MEMORY_DIR = original_dir
        mem.memory_cache.clear()
        shutil.rmtree(test_dir)


# ==================== Tests Bot (Response Cleaning) ====================

def test_clean_response_text():
    """Test le nettoyage de texte de réponse."""
    from core.response_handler import clean_response_text
    
    # Test nettoyage basique
    clean = clean_response_text("Salut ! Voici ma r\u00e9ponse.", None)
    assert clean == "Salut ! Voici ma r\u00e9ponse."
    
    # Test nettoyage avec JSON extrait
    clean = clean_response_text('Texte avant {"action": "NONE"} texte apr\u00e8s', '{"action": "NONE"}')
    assert "NONE" not in clean
    assert "avant" in clean
    assert "apr\u00e8s" in clean
    
    # Test nettoyage balises
    clean = clean_response_text("R\u00e9ponse <|channel|> normale", None)
    assert "<|channel|>" not in clean
    
    logger.info("OK Nettoyage r\u00e9ponse")


def test_parse_json_actions():
    """Test le parsing des actions JSON."""
    from core.response_handler import parse_json_actions
    
    # Test parsing valide
    actions, responses, is_none = parse_json_actions(
        '{"action": "KICK", "response": "Utilisateur kick\u00e9"}'
    )
    assert len(actions) == 1
    assert actions[0]["action"] == "KICK"
    assert "kick\u00e9" in responses[0]
    assert not is_none
    
    # Test parsing NONE
    actions, responses, is_none = parse_json_actions(
        '{"action": "NONE"}'
    )
    assert is_none
    
    # Test parsing tableau
    actions, responses, is_none = parse_json_actions(
        '[{"action": "A", "response": "R1"}, {"action": "B", "response": "R2"}]'
    )
    assert len(actions) == 2
    assert len(responses) == 2
    
    # Test parsing None
    actions, responses, is_none = parse_json_actions(None)
    assert actions == []
    assert responses == []
    assert not is_none
    
    logger.info("OK Parsing actions JSON")


# ==================== Tests Action Router ====================

def test_action_executor_init():
    """Test l'initialisation de l'ActionExecutor."""
    from core.action_executor import ActionExecutor
    from core.bot import Bot
    from core.llm import LLMManager
    import discord
    
    llm = LLMManager()
    intents = discord.Intents.default()
    bot = Bot(
        llama_model=llm, command_prefix="!", guild_id=123,
        system_prompt="Test", intents=intents
    )
    executor = ActionExecutor(bot)
    
    logger.info("✓ ActionExecutor initialisé OK")


# ==================== Tests Modération ====================

def test_moderation_db_init():
    """Test l'initialisation de la DB de modération."""
    import tempfile
    import brain.moderation as mod
    from brain import moderation
    
    original_path = mod.DB_PATH
    test_dir = tempfile.mkdtemp()
    mod.DB_PATH = os.path.join(test_dir, "test_mod.db")
    
    try:
        mod.init_db()
        
        # Vérifier que la table existe
        import sqlite3
        conn = sqlite3.connect(mod.DB_PATH)
        cursor = conn.cursor()
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='infractions'")
        assert cursor.fetchone() is not None
        conn.close()
        
        logger.info("✓ Initialisation DB modération OK")
    finally:
        mod.DB_PATH = original_path
        shutil.rmtree(test_dir)


# ==================== Point d'entrée ====================

if __name__ == "__main__":
    logger.info("=" * 50)
    logger.info("Tests Superviseur Bot")
    logger.info("=" * 50)
    
    tests = [
        test_imports,
        test_llm_manager_init,
        test_bot_init,
        test_extract_json_simple_object,
        test_extract_json_array,
        test_extract_json_with_comments,
        test_extract_json_none,
        test_extract_json_empty,
        test_extract_json_nested_braces,
        test_build_payload,
        test_memory_save_load,
        test_memory_add_interaction,
        test_memory_build_context,
        test_memory_delete,
        test_memory_extractive_summary,
        test_memory_flush_all,
        test_clean_response_text,
        test_parse_json_actions,
        test_action_executor_init,
        test_moderation_db_init,
    ]
    
    passed = 0
    failed = 0
    
    for test in tests:
        try:
            test()
            passed += 1
        except Exception as e:
            import traceback
            logger.error(f"✗ {test.__name__}: {e}")
            traceback.print_exc()
            failed += 1
    
    logger.info("=" * 50)
    logger.info(f"Résultats: {passed}/{passed + failed} tests réussis")
    logger.info("=" * 50)
    
    sys.exit(0 if failed == 0 else 1)