479 lines
No EOL
14 KiB
Python
479 lines
No EOL
14 KiB
Python
#!/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) |