tests/test_campaign_mcp.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Campagne de tests MCP complete - Lyra RAG v2
=============================================
Teste la detection outil + extraction arguments via _rule_based_detect.
Aucune execution de scripts reels.
Score:
PASS : outil correct + tous les args attendus presents
PARTIAL : outil correct + args obligatoires OK + args optionnels manquants
FAIL : mauvais outil ou arg obligatoire manquant
RULE_MISS: aucune regle ne matche (ira vers EPHAISTOS en production)
Usage:
python3 tests/test_campaign_mcp.py
"""
import sys
import unicodedata
from pathlib import Path
# Ajouter le repertoire racine au path
sys.path.insert(0, str(Path(__file__).parent.parent))
from lyra.core.pipeline import Pipeline
# ============================================================
# Format des cas de test
# (categorie, description, query, expected_tool,
# mandatory_args, optional_args)
#
# mandatory_args : dict - DOIT etre present (FAIL si absent)
# optional_args : dict - mentionne dans la requete mais
# peut ne pas etre extrait par les regles
# (PARTIAL si absent, pas FAIL)
# ============================================================
TESTS = [
# ================================================================
# FEDORA - vm_start
# ================================================================
("FEDORA/vm_start", "demarre basique",
"demarre preprod-01",
"fedora.vm_start", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_start", "lance avec nom",
"lance la vm test-server",
"fedora.vm_start", {"vm_name": "test-server"}, {}),
("FEDORA/vm_start", "boot",
"booter sandbox-02",
"fedora.vm_start", {"vm_name": "sandbox-02"}, {}),
("FEDORA/vm_start", "+ wait_ssh",
"demarre preprod-01 et attends le ssh",
"fedora.vm_start", {"vm_name": "preprod-01"}, {"wait_ssh": True}),
("FEDORA/vm_start", "+ wait_ip",
"demarre preprod-01 et attends son ip",
"fedora.vm_start", {"vm_name": "preprod-01"}, {"wait_ip": True}),
("FEDORA/vm_start", "alias start",
"start preprod-09",
"fedora.vm_start", {"vm_name": "preprod-09"}, {}),
# ================================================================
# FEDORA - vm_stop
# ================================================================
("FEDORA/vm_stop", "arrete basique",
"arrete preprod-01",
"fedora.vm_stop", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_stop", "stoppe",
"stoppe sandbox-02",
"fedora.vm_stop", {"vm_name": "sandbox-02"}, {}),
("FEDORA/vm_stop", "+ force",
"arrete de force preprod-01",
"fedora.vm_stop", {"vm_name": "preprod-01"}, {"force": True}),
("FEDORA/vm_stop", "stop alias",
"stop neutron-clone",
"fedora.vm_stop", {"vm_name": "neutron-clone"}, {}),
# ================================================================
# FEDORA - vm_destroy
# ================================================================
("FEDORA/vm_destroy", "supprime basique",
"supprime la vm sandbox-01",
"fedora.vm_destroy", {"vm_name": "sandbox-01"}, {}),
("FEDORA/vm_destroy", "detruit",
"detruis test-vm",
"fedora.vm_destroy", {"vm_name": "test-vm"}, {}),
("FEDORA/vm_destroy", "efface",
"efface la machine sandbox-02",
"fedora.vm_destroy", {"vm_name": "sandbox-02"}, {}),
("FEDORA/vm_destroy", "+ force",
"supprime de force sandbox-01",
"fedora.vm_destroy", {"vm_name": "sandbox-01"}, {"force": True}),
("FEDORA/vm_destroy", "+ keep_storage",
"supprime sandbox-01 mais garde le disque",
"fedora.vm_destroy", {"vm_name": "sandbox-01"}, {"keep_storage": True}),
# ================================================================
# FEDORA - vm_status
# ================================================================
("FEDORA/vm_status", "status specifique",
"status de preprod-01",
"fedora.vm_status", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_status", "etat VM",
"etat de la vm test-server",
"fedora.vm_status", {"vm_name": "test-server"}, {}),
("FEDORA/vm_status", "liste toutes VMs",
"liste mes vms",
"fedora.vm_status", {}, {}),
("FEDORA/vm_status", "affiche les machines",
"affiche toutes les machines virtuelles",
"fedora.vm_status", {}, {}),
("FEDORA/vm_status", "quelles sont mes VMs",
"quelles sont mes vms",
"fedora.vm_status", {}, {}),
("FEDORA/vm_status", "+ detailed",
"status detaille de preprod-01",
"fedora.vm_status", {"vm_name": "preprod-01"}, {"detailed": True}),
("FEDORA/vm_status", "donne moi les serveurs",
"donne moi la liste des serveurs",
"fedora.vm_status", {}, {}),
# ================================================================
# FEDORA - vm_exec
# ================================================================
("FEDORA/vm_exec", "execute commande",
"execute ls -la sur preprod-01",
"fedora.vm_exec", {"vm_name": "preprod-01", "command": "ls -la"}, {}),
("FEDORA/vm_exec", "execute uptime",
"execute uptime sur sandbox-02",
"fedora.vm_exec", {"vm_name": "sandbox-02", "command": "uptime"}, {}),
("FEDORA/vm_exec", "lance commande",
"lance la commande df -h sur test-server",
"fedora.vm_exec", {"vm_name": "test-server", "command": "df -h"}, {}),
("FEDORA/vm_exec", "+ sudo",
"execute systemctl restart nginx sur preprod-01 en sudo",
"fedora.vm_exec", {"vm_name": "preprod-01"}, {"sudo": True}),
("FEDORA/vm_exec", "+ user",
"execute whoami dans preprod-01",
"fedora.vm_exec", {"vm_name": "preprod-01", "command": "whoami"}, {}),
# ================================================================
# FEDORA - vm_copy
# ================================================================
("FEDORA/vm_copy", "copie fichier vers VM",
"copie /tmp/config.yaml vers preprod-01",
"fedora.vm_copy", {"vm_name": "preprod-01", "source": "/tmp/config.yaml"}, {}),
("FEDORA/vm_copy", "transfere vers VM",
"transfere /srv/user/script.sh sur sandbox-02",
"fedora.vm_copy", {"vm_name": "sandbox-02", "source": "/srv/user/script.sh"}, {}),
("FEDORA/vm_copy", "envoie fichier",
"envoie /etc/nginx.conf dans test-server",
"fedora.vm_copy", {"vm_name": "test-server", "source": "/etc/nginx.conf"}, {}),
("FEDORA/vm_copy", "+ recursive",
"copie /srv/user/projet vers preprod-01 en recursif",
"fedora.vm_copy", {"vm_name": "preprod-01"}, {"recursive": True}),
# ================================================================
# FEDORA - vm_snapshot
# ================================================================
("FEDORA/vm_snapshot", "cree snapshot",
"cree un snapshot de preprod-01",
"fedora.vm_snapshot", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_snapshot", "snapshot direct",
"snapshot sandbox-02",
"fedora.vm_snapshot", {"vm_name": "sandbox-02"}, {}),
("FEDORA/vm_snapshot", "instantane",
"instantane de test-server",
"fedora.vm_snapshot", {"vm_name": "test-server"}, {}),
("FEDORA/vm_snapshot", "capture",
"capture preprod-01",
"fedora.vm_snapshot", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_snapshot", "+ live",
"cree un snapshot live de preprod-01",
"fedora.vm_snapshot", {"vm_name": "preprod-01"}, {"live": True}),
# ================================================================
# FEDORA - vm_clone
# ================================================================
("FEDORA/vm_clone", "clone basique",
"clone preprod-01 en test-clone",
"fedora.vm_clone", {"source_vm": "preprod-01", "new_vm_name": "test-clone"}, {}),
("FEDORA/vm_clone", "duplique",
"duplique sandbox-02 en sandbox-03",
"fedora.vm_clone", {"source_vm": "sandbox-02", "new_vm_name": "sandbox-03"}, {}),
("FEDORA/vm_clone", "+ COW/linked",
"clone preprod-01 en test-cow en cow",
"fedora.vm_clone", {"source_vm": "preprod-01", "new_vm_name": "test-cow"}, {"linked": True}),
("FEDORA/vm_clone", "+ start apres",
"clone preprod-01 en test-clone et demarre",
"fedora.vm_clone", {"source_vm": "preprod-01", "new_vm_name": "test-clone"}, {"start": True}),
# ================================================================
# FEDORA - vm_clone_system
# ================================================================
("FEDORA/vm_clone_system", "clone systeme basique",
"clone systeme neutron en neutron-clone",
"fedora.vm_clone_system", {}, {}),
("FEDORA/vm_clone_system", "duplique systeme",
"duplique le systeme neutron en backup-system",
"fedora.vm_clone_system", {}, {}),
("FEDORA/vm_clone_system", "+ dry_run",
"clone systeme neutron en test en mode dry-run",
"fedora.vm_clone_system", {}, {"dry_run": True}),
# ================================================================
# FEDORA - vm_verify
# ================================================================
("FEDORA/vm_verify", "verifie VM specifique",
"verifie neutron-clone",
"fedora.vm_verify", {"vm_name": "neutron-clone"}, {}),
("FEDORA/vm_verify", "verifie la VM",
"verifie la vm preprod-01",
"fedora.vm_verify", {"vm_name": "preprod-01"}, {}),
("FEDORA/vm_verify", "verif alias",
"verif sandbox-02",
"fedora.vm_verify", {"vm_name": "sandbox-02"}, {}),
("FEDORA/vm_verify", "+ quick",
"verifie rapidement neutron-clone",
"fedora.vm_verify", {"vm_name": "neutron-clone"}, {"quick": True}),
("FEDORA/vm_verify", "+ verbose",
"verifie neutron-clone en detail",
"fedora.vm_verify", {"vm_name": "neutron-clone"}, {"verbose": True}),
("FEDORA/vm_verify", "+ compare_packages",
"verifie neutron-clone avec comparaison des paquets",
"fedora.vm_verify", {"vm_name": "neutron-clone"}, {"compare_packages": True}),
# ================================================================
# FEDORA - vm_export
# ================================================================
("FEDORA/vm_export", "export classic defaut",
"exporte preprod-01",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "classic"}, {}),
("FEDORA/vm_export", "export mode exam",
"exporte preprod-01 en mode exam",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "exam"}, {}),
("FEDORA/vm_export", "export mode examen",
"exporte system-clone-final mode examen",
"fedora.vm_export", {"vm_name": "system-clone-final", "mode": "exam"}, {}),
("FEDORA/vm_export", "export mode custom",
"exporte preprod-01 en mode custom",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "custom"}, {}),
("FEDORA/vm_export", "export personnalise",
"exporte preprod-01 en mode personnalise",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "custom"}, {}),
("FEDORA/vm_export", "emballe VM",
"emballe sandbox-02",
"fedora.vm_export", {"vm_name": "sandbox-02", "mode": "classic"}, {}),
("FEDORA/vm_export", "+ output_path",
"exporte preprod-01 vers /mnt/exports",
"fedora.vm_export", {"vm_name": "preprod-01"}, {"output_path": "/mnt/exports"}),
("FEDORA/vm_export", "+ force",
"exporte de force preprod-01",
"fedora.vm_export", {"vm_name": "preprod-01"}, {"force": True}),
("FEDORA/vm_export", "+ dry_run",
"exporte preprod-01 en mode test dry-run",
"fedora.vm_export", {"vm_name": "preprod-01"}, {"dry_run": True}),
# ================================================================
# FEDORA - vm_import
# ================================================================
("FEDORA/vm_import", "importe archive",
"importe /srv/user/vm-exports/preprod-01-export-20260227-classic.tar.gz",
"fedora.vm_import", {"archive_path": "/srv/user/vm-exports/preprod-01-export-20260227-classic.tar.gz"}, {}),
("FEDORA/vm_import", "importe avec nouveau nom",
"importe /srv/user/vm-exports/preprod-01-export-20260227-classic.tar.gz sous le nom test-import",
"fedora.vm_import",
{"archive_path": "/srv/user/vm-exports/preprod-01-export-20260227-classic.tar.gz", "new_name": "test-import"},
{}),
("FEDORA/vm_import", "charge VM",
"charge la vm depuis ~/vm-exports/system-clone-final-export-20260226-211346-classic.tar.gz",
"fedora.vm_import", {"archive_path": "~/vm-exports/system-clone-final-export-20260226-211346-classic.tar.gz"}, {}),
("FEDORA/vm_import", "+ start apres import",
"importe /tmp/test.tar.gz et demarre",
"fedora.vm_import", {"archive_path": "/tmp/test.tar.gz"}, {"start": True}),
("FEDORA/vm_import", "+ dry_run",
"importe /tmp/test.tar.gz en dry-run",
"fedora.vm_import", {"archive_path": "/tmp/test.tar.gz"}, {"dry_run": True}),
# ================================================================
# BACKUP - backup_status
# ================================================================
("BACKUP/status", "status backup",
"status des backups",
"fedora.backup_status", {}, {}),
("BACKUP/status", "etat backup",
"etat des sauvegardes",
"fedora.backup_status", {}, {}),
("BACKUP/status", "dashboard backup",
"dashboard backup",
"fedora.backup_status", {}, {}),
("BACKUP/status", "status sauvegarde",
"status de mes sauvegardes",
"fedora.backup_status", {}, {}),
# ================================================================
# BACKUP - backup_list
# ================================================================
("BACKUP/list", "liste backups",
"liste les backups",
"fedora.backup_list", {}, {}),
("BACKUP/list", "affiche sauvegardes",
"affiche mes sauvegardes",
"fedora.backup_list", {}, {}),
("BACKUP/list", "montre backups",
"montre les backups",
"fedora.backup_list", {}, {}),
("BACKUP/list", "quelles sauvegardes",
"quelles sont mes sauvegardes",
"fedora.backup_list", {}, {}),
("BACKUP/list", "+ type timeshift",
"liste les backups timeshift",
"fedora.backup_list", {}, {"type": "timeshift"}),
("BACKUP/list", "+ type borg",
"liste les sauvegardes borg",
"fedora.backup_list", {}, {"type": "borg"}),
("BACKUP/list", "+ detailed",
"liste les backups en detail",
"fedora.backup_list", {}, {"detailed": True}),
# ================================================================
# BACKUP - backup_create
# ================================================================
("BACKUP/create", "cree backup",
"cree un backup",
"fedora.backup_create", {}, {}),
("BACKUP/create", "fais sauvegarde",
"fais une sauvegarde",
"fedora.backup_create", {}, {}),
("BACKUP/create", "lance backup",
"lance un backup",
"fedora.backup_create", {}, {}),
("BACKUP/create", "backup de VM",
"cree un backup de preprod-01",
"fedora.backup_create", {}, {}),
("BACKUP/create", "+ type timeshift",
"cree un backup timeshift",
"fedora.backup_create", {}, {"type": "timeshift"}),
("BACKUP/create", "+ verify",
"cree un backup et verifie",
"fedora.backup_create", {}, {"verify": True}),
("BACKUP/create", "+ dry_run",
"cree un backup en dry-run",
"fedora.backup_create", {}, {"dry_run": True}),
# ================================================================
# BACKUP - backup_restore
# ================================================================
("BACKUP/restore", "restaure backup",
"restaure le backup",
"fedora.backup_restore", {}, {}),
("BACKUP/restore", "restaure sauvegarde",
"restaure ma sauvegarde",
"fedora.backup_restore", {}, {}),
("BACKUP/restore", "restore alias",
"restore le backup de preprod-01",
"fedora.backup_restore", {}, {}),
("BACKUP/restore", "+ dry_run",
"restaure le backup en dry-run",
"fedora.backup_restore", {}, {"dry_run": True}),
("BACKUP/restore", "+ force",
"restaure de force la sauvegarde",
"fedora.backup_restore", {}, {"force": True}),
# ================================================================
# BACKUP - backup_verify
# ================================================================
("BACKUP/verify", "verifie backup",
"verifie le backup",
"fedora.backup_verify", {}, {}),
("BACKUP/verify", "verifie sauvegarde",
"verifie la sauvegarde",
"fedora.backup_verify", {}, {}),
("BACKUP/verify", "controle backup",
"controle le backup",
"fedora.backup_verify", {}, {}),
("BACKUP/verify", "verifie backup VM",
"verifie le backup de preprod-01",
"fedora.backup_verify", {}, {}),
("BACKUP/verify", "+ deep",
"verifie en profondeur le backup",
"fedora.backup_verify", {}, {"deep": True}),
("BACKUP/verify", "+ quick",
"verifie rapidement le backup",
"fedora.backup_verify", {}, {"quick": True}),
# ================================================================
# BACKUP - backup_clean
# ================================================================
("BACKUP/clean", "nettoie backups",
"nettoie les backups",
"fedora.backup_clean", {}, {}),
("BACKUP/clean", "purge sauvegardes",
"purge les sauvegardes",
"fedora.backup_clean", {}, {}),
("BACKUP/clean", "supprime anciens backups",
"supprime les anciens backups",
"fedora.backup_clean", {}, {}),
("BACKUP/clean", "efface backup",
"efface les vieux backups",
"fedora.backup_clean", {}, {}),
("BACKUP/clean", "+ dry_run",
"nettoie les backups en dry-run",
"fedora.backup_clean", {}, {"dry_run": True}),
("BACKUP/clean", "+ keep_last",
"nettoie les backups garde les 3 derniers",
"fedora.backup_clean", {}, {"keep_last": 3}),
# ================================================================
# DENON - power
# ================================================================
("DENON/power", "allume denon",
"allume le denon",
"denon.power_on", {}, {}),
("DENON/power", "demarre denon",
"demarre le denon",
"denon.power_on", {}, {}),
("DENON/power", "eteins denon",
"eteins le denon",
"denon.power_off", {}, {}),
("DENON/power", "arrete denon",
"arrete le denon",
"denon.power_off", {}, {}),
("DENON/power", "stop denon",
"stop le denon",
"denon.power_off", {}, {}),
# ================================================================
# DENON - volume
# ================================================================
("DENON/volume", "volume denon 44",
"volume denon a 44",
"denon.volume_set", {"level": 44}, {}),
("DENON/volume", "volume denon 50",
"mets le volume denon a 50",
"denon.volume_set", {"level": 50}, {}),
("DENON/volume", "monte volume denon",
"monte le volume denon",
"denon.volume_up", {}, {}),
("DENON/volume", "baisse volume denon",
"baisse le volume denon",
"denon.volume_down", {}, {}),
("DENON/volume", "50 de volume denon",
"mets 50 de volume sur le denon",
"denon.volume_set", {"level": 50}, {}),
# ================================================================
# DENON - mute
# ================================================================
("DENON/mute", "mute denon",
"mute le denon",
"denon.mute_on", {}, {}),
("DENON/mute", "sourdine denon",
"active la sourdine sur le denon",
"denon.mute_on", {}, {}),
("DENON/mute", "coupe son denon",
"coupe le son du denon",
"denon.mute_on", {}, {}),
("DENON/mute", "demute denon",
"demute le denon",
"denon.mute_off", {}, {}),
("DENON/mute", "unmute denon",
"unmute le denon",
"denon.mute_off", {}, {}),
("DENON/mute", "desactive mute denon",
"desactive le mute sur le denon",
"denon.mute_off", {}, {}),
# ================================================================
# TV - power (RULE_MISS expected - pas dans les regles)
# ================================================================
("TV/power", "allume TV",
"allume la tv",
"tv.power_on", {}, {}),
("TV/power", "eteins TV",
"eteins la tele",
"tv.power_off", {}, {}),
("TV/power", "TV en veille",
"mets la tv en veille",
"tv.power_off", {}, {}),
# ================================================================
# TV - volume (redirige vers DENON si HDMI ARC)
# ================================================================
("TV/volume", "monte volume TV",
"monte le volume de la tv",
"tv.volume_up", {}, {}),
("TV/volume", "baisse volume TV",
"baisse le son de la tele",
"tv.volume_down", {}, {}),
("TV/volume", "volume TV 40",
"mets le volume de la tv a 40",
"tv.volume_set", {}, {}),
("TV/volume", "mute TV",
"mets la tv en muet",
"tv.mute", {}, {}),
# ================================================================
# TV - applications
# ================================================================
("TV/app", "lance Netflix sur TV",
"lance netflix sur la tv",
"tv.launch_app", {}, {}),
("TV/app", "ouvre YouTube TV",
"ouvre youtube sur la tele",
"tv.launch_app", {}, {}),
("TV/app", "youtube video",
"joue cette video youtube sur la tv https://youtu.be/dQw4w9WgXcQ",
"tv.youtube_video", {}, {}),
# ================================================================
# TV - ambilight
# ================================================================
("TV/ambilight", "allume ambilight",
"allume l ambilight",
"tv.ambilight_on", {}, {}),
("TV/ambilight", "eteins ambilight",
"eteins l ambilight",
"tv.ambilight_off", {}, {}),
("TV/ambilight", "ambilight couleur",
"mets l ambilight en bleu",
"tv.ambilight_color", {}, {}),
# ================================================================
# HUE - turn_on_group (dans les regles)
# ================================================================
("HUE/group", "allume lumieres",
"allume les lumieres",
"hue.turn_on_group", {"group_id": 81}, {}),
("HUE/group", "allume toutes lumieres",
"allume toutes les lumieres",
"hue.turn_on_group", {"group_id": 81}, {}),
("HUE/group", "eteins lumieres groupe",
"eteins les lumieres",
"hue.turn_off_group", {"group_id": 81}, {}),
# ================================================================
# HUE - turn_on/off_light (dans les regles pour off)
# ================================================================
("HUE/light", "eteins lumiere bureau",
"eteins la lumiere bureau",
"hue.turn_off_light", {}, {}),
("HUE/light", "eteins lumiere salon",
"eteins la lumiere salon",
"hue.turn_off_light", {}, {}),
("HUE/light", "allume lumiere chevet",
"allume la lumiere chevet",
"hue.turn_on_light", {}, {}), # RULE_MISS
# ================================================================
# HUE - brightness / color (RULE_MISS)
# ================================================================
("HUE/brightness", "luminosite 50%",
"mets la luminosite a 50 pour cent",
"hue.set_group_brightness", {}, {}),
("HUE/brightness", "lumiere plus forte",
"mets les lumieres plus fortes",
"hue.set_group_brightness", {}, {}),
("HUE/color", "couleur rouge",
"mets les lumieres en rouge",
"hue.set_group_color_rgb", {}, {}),
("HUE/color", "ambiance bleue",
"mets une ambiance bleue",
"hue.set_group_color_rgb", {}, {}),
# ================================================================
# HUE - scene
# ================================================================
("HUE/scene", "active scene relax",
"active la scene relax",
"hue.activate_scene_by_name", {}, {}),
("HUE/scene", "scene lecture",
"lance la scene lecture",
"hue.activate_scene_by_name", {}, {}),
# ================================================================
# Edge cases - collisions et cas limites
# ================================================================
("EDGE/collision", "start ne matche pas TV",
"lance netflix sur la tv",
"tv.launch_app", {}, {}), # ne doit PAS declencher vm_start
("EDGE/collision", "stop ne matche pas backup",
"status des backups",
"fedora.backup_status", {}, {}),
("EDGE/collision", "verifie backup vs vm",
"verifie le backup de preprod-01",
"fedora.backup_verify", {}, {}), # backup_verify avant vm_verify
("EDGE/collision", "supprime backups vs vm",
"supprime les anciens backups",
"fedora.backup_clean", {}, {}), # backup_clean avant vm_destroy
("EDGE/collision", "clone systeme vs clone normal",
"clone systeme neutron en backup",
"fedora.vm_clone_system", {}, {}), # vm_clone_system avant vm_clone
("EDGE/args", "export mode exam syntaxe alternative",
"exporte system-clone-final en mode examen",
"fedora.vm_export", {"vm_name": "system-clone-final", "mode": "exam"}, {}),
("EDGE/args", "import avec renommage",
"importe /srv/user/vm-exports/test.tar.gz sous le nom ma-vm",
"fedora.vm_import", {"new_name": "ma-vm"}, {}),
("EDGE/args", "vm_status global vs specifique",
"status de preprod-01",
"fedora.vm_status", {"vm_name": "preprod-01"}, {}),
("EDGE/args", "snapshot sans nom VM",
"cree un snapshot",
None, {}, {}), # doit demander clarification
("EDGE/args", "copie chemin complexe",
"copie /srv/user/projet/config.json vers sandbox-02",
"fedora.vm_copy", {"source": "/srv/user/projet/config.json", "vm_name": "sandbox-02"}, {}),
("EDGE/args", "exec commande avec espaces",
"execute apt-get update sur preprod-01",
"fedora.vm_exec", {"vm_name": "preprod-01", "command": "apt-get update"}, {}),
("EDGE/args", "export custom synonyme sur-mesure",
"exporte preprod-01 en mode sur-mesure",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "custom"}, {}),
("EDGE/args", "export custom synonyme choix",
"exporte preprod-01 avec choix des operations",
"fedora.vm_export", {"vm_name": "preprod-01", "mode": "custom"}, {}),
]
# ============================================================
# Runner
# ============================================================
def normalize(q):
"""Normalise la query comme le fait _rule_based_detect."""
q = unicodedata.normalize('NFD', q.lower())
return ''.join(c for c in q if unicodedata.category(c) != 'Mn')
def tool_matches(result_tool, expected_tool):
"""Compare les noms d'outils avec ou sans prefixe serveur."""
if expected_tool is None:
return result_tool is None
if result_tool is None:
return False
r = result_tool.split(".")[-1] if "." in result_tool else result_tool
e = expected_tool.split(".")[-1] if "." in expected_tool else expected_tool
return r == e
def check_args(result_args, mandatory, optional):
"""Verifie les arguments. Retourne (mandatory_ok, missing_optional)."""
if result_args is None:
result_args = {}
missing_mandatory = []
for k, v in mandatory.items():
rv = result_args.get(k)
if rv is None:
missing_mandatory.append(k)
elif v != "" and str(rv) != str(v):
missing_mandatory.append(f"{k}={v}(got={rv})")
missing_optional = [k for k in optional if k not in result_args]
return missing_mandatory, missing_optional
def run_tests():
categories = {}
results = []
PASS = "PASS"
PARTIAL = "PARTIAL"
FAIL = "FAIL"
RULE_MISS = "RULE_MISS"
for cat, desc, query, expected_tool, mandatory_args, optional_args in TESTS:
raw_q = normalize(query)
analysis = Pipeline._rule_based_detect(raw_q)
if analysis is not None:
analysis = Pipeline._enrich_optional_args(raw_q, analysis)
result_tool = analysis.tool if analysis else None
result_args = analysis.arguments if analysis else {}
tool_ok = tool_matches(result_tool, expected_tool)
missing_mand, missing_opt = check_args(result_args, mandatory_args, optional_args)
if expected_tool is None:
# On attend que rien ne matche (clarification requise)
status = PASS if result_tool is None else FAIL
elif result_tool is None:
status = RULE_MISS
elif not tool_ok:
status = FAIL
elif missing_mand:
status = FAIL
elif missing_opt:
status = PARTIAL
else:
status = PASS
entry = {
"cat": cat, "desc": desc, "query": query,
"expected_tool": expected_tool, "result_tool": result_tool,
"result_args": result_args, "mandatory_args": mandatory_args,
"missing_mand": missing_mand, "missing_opt": missing_opt,
"status": status
}
results.append(entry)
categories.setdefault(cat.split("/")[0], []).append(entry)
return results, categories
def print_report(results, categories):
G = "\033[32m"
Y = "\033[33m"
R = "\033[31m"
B = "\033[34m"
BOLD = "\033[1m"
DIM = "\033[2m"
RESET = "\033[0m"
STATUS_COLOR = {
"PASS": G, "PARTIAL": Y, "FAIL": R, "RULE_MISS": B
}
print(f"\n{BOLD}{'='*70}{RESET}")
print(f"{BOLD} CAMPAGNE DE TESTS MCP - LYRA RAG v2{RESET}")
print(f"{BOLD}{'='*70}{RESET}\n")
# Detail par test
current_cat = None
for r in results:
cat_group = r["cat"].split("/")[0]
if cat_group != current_cat:
current_cat = cat_group
print(f"\n{BOLD}--- {cat_group} ---{RESET}")
sc = r["status"]
color = STATUS_COLOR[sc]
pad = r["desc"][:35].ljust(36)
tool_str = (r["result_tool"] or "NONE").split(".")[-1][:18].ljust(19)
args_str = ""
if r["missing_mand"]:
args_str = f" {R}ARGS_MANQUANTS={r['missing_mand']}{RESET}"
elif r["missing_opt"]:
args_str = f" {Y}OPT_MISS={r['missing_opt']}{RESET}"
print(f" {color}[{sc:<9}]{RESET} {DIM}{pad}{RESET} -> {tool_str}{args_str}")
# Score par categorie
print(f"\n{BOLD}{'='*70}{RESET}")
print(f"{BOLD} SCORES PAR CATEGORIE{RESET}")
print(f"{BOLD}{'='*70}{RESET}")
total_pass = total_partial = total_fail = total_miss = 0
for cat_name, entries in sorted(categories.items()):
n = len(entries)
p = sum(1 for e in entries if e["status"] == "PASS")
pa = sum(1 for e in entries if e["status"] == "PARTIAL")
f = sum(1 for e in entries if e["status"] == "FAIL")
m = sum(1 for e in entries if e["status"] == "RULE_MISS")
total_pass += p
total_partial += pa
total_fail += f
total_miss += m
score_pct = round(100 * (p + 0.5 * pa) / n) if n else 0
bar = f"{G}{'#' * (p)}{Y}{'~' * pa}{B}{'?' * m}{R}{'-' * f}{RESET}"
print(f" {BOLD}{cat_name:<8}{RESET} {bar} "
f"{G}{p}P{RESET} {Y}{pa}~{RESET} {B}{m}?{RESET} {R}{f}F{RESET} "
f"/ {n} [{score_pct}%]")
total = len(results)
score_final = round(100 * (total_pass + 0.5 * total_partial) / total)
print(f"\n{BOLD}{'='*70}{RESET}")
print(f"{BOLD} SCORE FINAL : {G}{total_pass}{RESET}{BOLD} PASS "
f"{Y}{total_partial}{RESET}{BOLD} PARTIAL "
f"{B}{total_miss}{RESET}{BOLD} RULE_MISS "
f"{R}{total_fail}{RESET}{BOLD} FAIL / {total} tests{RESET}")
print(f"{BOLD} POURCENTAGE : {score_final}%{RESET} "
f"{DIM}(PASS=1pt, PARTIAL=0.5pt, RULE_MISS/FAIL=0pt){RESET}")
print(f"\n{BOLD} LEGENDE:{RESET}")
print(f" {G}[PASS] {RESET} Outil correct + tous les args attendus")
print(f" {Y}[PARTIAL] {RESET} Outil correct + args obligatoires OK + args optionnels manquants")
print(f" {B}[RULE_MISS]{RESET} Aucune regle ne matche -> ira vers EPHAISTOS en production")
print(f" {R}[FAIL] {RESET} Mauvais outil ou arg obligatoire manquant\n")
# Rapport texte brut pour PDF/NotebookLM
return generate_text_report(results, categories, total_pass, total_partial, total_fail, total_miss, total, score_final)
def generate_text_report(results, categories, total_pass, total_partial, total_fail, total_miss, total, score_final):
lines = []
lines.append("CAMPAGNE DE TESTS MCP - LYRA RAG v2")
lines.append("=" * 70)
lines.append("")
lines.append("OBJECTIF")
lines.append("--------")
lines.append("Tester la detection outil + extraction arguments du pipeline Lyra")
lines.append("via _rule_based_detect(), sans executer aucun script reel.")
lines.append("")
lines.append("LEGENDE")
lines.append("-------")
lines.append("PASS : Outil correct + tous les args attendus presens")
lines.append("PARTIAL : Outil correct + args obligatoires OK + args optionnels manquants")
lines.append("RULE_MISS : Aucune regle ne matche (ira vers EPHAISTOS en production)")
lines.append("FAIL : Mauvais outil ou arg obligatoire manquant")
lines.append("")
lines.append("RESULTATS DETAILLES")
lines.append("-------------------")
current_cat = None
for r in results:
cat_group = r["cat"].split("/")[0]
if cat_group != current_cat:
current_cat = cat_group
lines.append("")
lines.append(f"=== {cat_group} ===")
status = r["status"]
desc = r["desc"]
query = r["query"]
result_tool = r["result_tool"] or "NONE"
expected_tool = r["expected_tool"] or "NONE"
result_args = r["result_args"]
missing_mand = r["missing_mand"]
missing_opt = r["missing_opt"]
lines.append(f"[{status:<9}] {desc}")
lines.append(f" Requete : \"{query}\"")
lines.append(f" Attendu : {expected_tool}")
lines.append(f" Obtenu : {result_tool}")
if result_args:
lines.append(f" Args : {result_args}")
if missing_mand:
lines.append(f" FAIL args : {missing_mand}")
if missing_opt:
lines.append(f" PARTIAL : args optionnels manquants : {missing_opt}")
lines.append("")
lines.append("SCORES PAR CATEGORIE")
lines.append("--------------------")
for cat_name, entries in sorted(categories.items()):
n = len(entries)
p = sum(1 for e in entries if e["status"] == "PASS")
pa = sum(1 for e in entries if e["status"] == "PARTIAL")
f = sum(1 for e in entries if e["status"] == "FAIL")
m = sum(1 for e in entries if e["status"] == "RULE_MISS")
score_pct = round(100 * (p + 0.5 * pa) / n) if n else 0
lines.append(f" {cat_name:<10} : {p} PASS {pa} PARTIAL {m} RULE_MISS {f} FAIL / {n} tests [{score_pct}%]")
lines.append("")
lines.append("SCORE FINAL")
lines.append("-----------")
lines.append(f" PASS : {total_pass}")
lines.append(f" PARTIAL : {total_partial}")
lines.append(f" RULE_MISS : {total_miss}")
lines.append(f" FAIL : {total_fail}")
lines.append(f" TOTAL : {total} tests")
lines.append(f" SCORE : {score_final}%")
lines.append("")
lines.append("ANALYSE")
lines.append("-------")
lines.append("Les RULE_MISS correspondent aux outils TV, HUE (sauf turn_on/off_group/light)")
lines.append("qui ne sont pas couverts par _rule_based_detect.")
lines.append("Ces requetes sont traitees par EPHAISTOS (LLM) en production.")
lines.append("Les PARTIAL indiquent des args optionnels mentionnes dans la requete mais non")
lines.append("extraits par les regles (attendu - les regles n'extraient pas tous les params).")
return "\n".join(lines)
def resume_json(results, categories) -> dict:
"""Resume chiffre de la campagne, destine a benchmarks/results/.
Cette campagne n'interroge aucun LLM : elle mesure les regles seules
(Pipeline._rule_based_detect), donc elle est deterministe et rejouable.
"""
from collections import Counter
statuts = Counter(r["status"] for r in results)
par_cat = {
cat: dict(Counter(r["status"] for r in entries))
for cat, entries in sorted(categories.items())
}
total = len(results)
return {
"banc": "regles",
"cas": total,
"statuts": dict(statuts),
"taux_pass": round(statuts["PASS"] / total, 4) if total else 0.0,
"par_categorie": par_cat,
"echecs": [
{"query": r["query"], "attendu": r["expected_tool"], "obtenu": r["result_tool"],
"statut": r["status"]}
for r in results if r["status"] in ("FAIL", "RULE_MISS")
],
}
if __name__ == "__main__":
results, categories = run_tests()
report_text = print_report(results, categories)
# Sauvegarder le rapport texte
report_path = Path(__file__).parent / "test_campaign_report.txt"
report_path.write_text(report_text, encoding="utf-8")
print(f"Rapport sauvegarde : {report_path}")
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from bench_common import ecrire_resultat # noqa: E402
chemin = ecrire_resultat("regles", resume_json(results, categories))
print(f"Resume publie : {chemin}")