tests/test_campaign_oneshot.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# NOTE: lancer avec le venv lyra: .venv/bin/python tests/test_campaign_oneshot.py
"""
Campagne de tests ONE-SHOT - Lyra RAG v2
=========================================
Pour chaque test du campaign MCP, execute les deux phases:
Phase 1 (DRY-RUN) : _rule_based_detect uniquement (sans Ollama, <1s total)
Phase 2 (ONE-SHOT) : pipeline complet avec Ollama + RAG + regles + EPHAISTOS
(init unique ~5-10s, puis ~1-3s par requete)
Avantages vs test_campaign_mcp.py :
- Detecte les bugs pipeline qui passent les regles mais echouent en production
- Montre les divergences dry-run vs realite (ex: IntentClassifier qui classe mal)
- Source de chaque decision : "rule" (sans LLM) ou "ephaistos" (Qwen)
Usage:
python3 tests/test_campaign_oneshot.py # Tous les tests
python3 tests/test_campaign_oneshot.py --category FEDORA # Une categorie
python3 tests/test_campaign_oneshot.py --fails-only # Affiche seulement les echecs
python3 tests/test_campaign_oneshot.py --dry-only # Phase 1 uniquement (rapide)
python3 tests/test_campaign_oneshot.py --oneshot-only # Phase 2 uniquement
Notes:
- Le pipeline est initialise une seule fois (Ollama, ChromaDB, SentenceTransformer, MCP)
- Aucune action MCP n'est executee : pipeline.process() retourne la decision sans executer
- Requiert Ollama actif avec les modeles configures dans config.yaml
"""
import argparse
import sys
import time
from pathlib import Path
from typing import Optional
# --- Path setup ---
ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(Path(__file__).parent))
from test_campaign_mcp import TESTS, check_args, normalize, tool_matches
# --- Constantes statut ---
PASS = "PASS"
PARTIAL = "PARTIAL"
FAIL = "FAIL"
RULE_MISS = "RULE_MISS"
ERROR = "ERROR" # Exception pendant le one-shot
# --- Evaluation ---
def eval_result(result_tool, result_args, result_pending, expected_tool, mandatory_args, optional_args):
"""Evalue un resultat de test (dry-run ou one-shot)."""
# Combiner args connus + pending comme "non extrait" pour les mandatory checks
all_args = dict(result_args or {})
if expected_tool is None:
# Aucune action attendue : soit le pipeline ne propose rien, soit il
# propose un outil en ATTENTE d'un argument obligatoire (le pipeline
# pose alors une question, rien ne s'execute). "cree un snapshot" sans
# nom de VM doit demander la VM ; c'est une clarification, pas une
# action. Un outil propose sans argument en attente reste un echec.
if result_tool is None:
return PASS
return PASS if result_pending else FAIL
if result_tool is None:
return RULE_MISS
if not tool_matches(result_tool, expected_tool):
return FAIL
missing_mand, missing_opt = check_args(all_args, mandatory_args, optional_args)
# Aussi FAIL si un mandatory_arg est dans pending (pas extrait de la requete)
for k in list(mandatory_args.keys()):
if k in (result_pending or []) and k not in [m.split("=")[0] for m in missing_mand]:
missing_mand.append(f"{k}(pending)")
if missing_mand:
return FAIL
if missing_opt:
return PARTIAL
return PASS
# --- Phase 1 : DRY-RUN ---
def run_dryrun_phase():
"""Execute la phase dry-run pour tous les tests.
Appelle Pipeline._rule_based_detect + _enrich_optional_args
sans initialiser Ollama ni ChromaDB.
Returns:
list[dict]: resultats bruts par test
"""
from lyra.core.pipeline import Pipeline
results = []
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 {}
result_pending = analysis.missing_args if analysis else []
reasoning = (analysis.reasoning or "") if analysis else ""
status = eval_result(result_tool, result_args, result_pending,
expected_tool, mandatory_args, optional_args)
results.append({
"cat": cat, "desc": desc, "query": query,
"expected_tool": expected_tool,
"result_tool": result_tool,
"result_args": result_args,
"result_pending": result_pending,
"status": status,
"reasoning": reasoning,
"source": "rule" if reasoning.startswith("rule:") else ("ephaistos" if result_tool else "no_match"),
})
return results
# --- Phase 2 : ONE-SHOT ---
# Noms de VMs fictifs utilises dans les tests (non existants sur l'hote)
_TEST_VM_NAMES = {
"preprod-01", "preprod-09",
"sandbox-01", "sandbox-02",
# sandbox-03, test-clone, test-cow, neutron-clone, backup-system sont des
# DESTINATIONS de clone/export -- elles ne doivent PAS exister pour eviter
# que le workflow vm_clone declenche le cas "nom deja pris"
"test-server", "test-vm",
"neutron",
"system-clone-final", "prod-clone",
"electron-backup-test",
"arch-base", "fedora-base", "ubuntu-base", "neutron-template-05",
}
def _mock_vm_calls(pipeline):
"""Patche les appels MCP reels du pipeline pour les tests.
Remplace les methodes et fonctions standalone qui interrogent les VMs
reelles par des stubs previsibles.
Les VMs fictives des tests sont toutes considerees comme existantes
et eteintes.
"""
import types
import lyra.core.workflows.vm_clone as _wf_clone
import lyra.core.workflows.vm_snapshot as _wf_snapshot
v2 = pipeline._pipeline_v2 if hasattr(pipeline, "_pipeline_v2") else pipeline
# _validate_vm_existence: toujours OK (pas d'erreur "VM inexistante")
v2._validate_vm_existence = types.MethodType(
lambda self, analysis, query: None, v2
)
# _get_vm_state: VM fictive = eteinte, pas d'IP
v2._get_vm_state = types.MethodType(
lambda self, vm_name: {"running": False, "ip": None}, v2
)
# _get_existing_vm_names: retourne le set des VMs de test
v2._get_existing_vm_names = types.MethodType(
lambda self: list(_TEST_VM_NAMES), v2
)
# Patch les fonctions standalone importees par les workflows (from ..validation import ...)
# Ces references sont dans l'espace de noms du module workflow, pas celui du pipeline
_wf_clone.get_existing_vm_names = lambda hestia: list(_TEST_VM_NAMES)
_wf_clone.get_vm_state = lambda hestia, vm_name: {"running": False, "ip": None}
_wf_snapshot.get_existing_vm_names = lambda hestia: list(_TEST_VM_NAMES)
def init_pipeline(config_path=None):
"""Initialise le pipeline complet (Ollama + RAG + MCP).
Les appels MCP reels (validation VM, etat) sont patched pour
que les tests fonctionnent sans infrastructure reelle.
Returns:
pipeline instance prete a l'emploi
Raises:
Exception si l'init echoue (modele manquant, etc.)
"""
import yaml
from lyra.core.config import RAGConfig
config_path = Path(config_path) if config_path else ROOT / "config.yaml"
config = RAGConfig.from_yaml(config_path)
# Lire la config enhanced depuis YAML
with open(config_path) as f:
cfg = yaml.safe_load(f)
enhanced_raw = cfg.get("rag_enhanced", {})
# Toujours utiliser EnhancedPipeline (c'est le mode par defaut de run.sh)
from lyra.rag_enhanced import EnhancedPipeline
from lyra.rag_enhanced.config import RAGEnhancedConfig
enhanced_config = RAGEnhancedConfig.from_dict(enhanced_raw)
pipeline = EnhancedPipeline(
config=config,
enhanced_config=enhanced_config,
enabled=enhanced_config.enabled,
tts_mode=False
)
pipeline.initialize()
_mock_vm_calls(pipeline)
return pipeline
def run_oneshot_single(pipeline, query: str, expected_tool: str,
mandatory_args: dict, optional_args: dict) -> dict:
"""Traite une requete via le pipeline complet.
Args:
pipeline: Pipeline initialise
query: Requete utilisateur (non normalisee, telle quelle)
expected_tool: Outil attendu
mandatory_args: Args obligatoires
optional_args: Args optionnels
Returns:
dict avec status, result_tool, result_args, source, etc.
"""
from lyra.core.types import QueryType
# Reset session avant chaque test pour eviter la contamination
# (les workflows interactifs laissent pending_action/pending_choice en session)
v2 = pipeline._pipeline_v2 if hasattr(pipeline, "_pipeline_v2") else pipeline
v2._session.clear()
t0 = time.perf_counter()
try:
result = pipeline.process(query)
except Exception as exc:
elapsed = time.perf_counter() - t0
return {
"result_tool": None, "result_args": {}, "result_pending": [],
"status": ERROR, "source": "error",
"reasoning": str(exc), "elapsed": elapsed,
"error": str(exc),
}
elapsed = time.perf_counter() - t0
# Extraire les infos du resultat
if result.tool_call:
result_tool = result.tool_call.get("name")
result_args = result.tool_call.get("arguments", {})
else:
result_tool = None
result_args = {}
result_pending = getattr(result, "pending_args", []) or []
meta = getattr(result, "analysis_meta", None) or {}
source = meta.get("source", "unknown")
reasoning = meta.get("reasoning", "")
# Si le pipeline a classifie comme KNOWLEDGE alors qu'on attend une action
if result.query_type == QueryType.KNOWLEDGE and expected_tool is not None:
return {
"result_tool": None, "result_args": {}, "result_pending": [],
"status": FAIL, "source": source,
"reasoning": "IntentClassifier: KNOWLEDGE (attendu ACTION)",
"elapsed": elapsed, "error": None,
}
status = eval_result(result_tool, result_args, result_pending,
expected_tool, mandatory_args, optional_args)
return {
"result_tool": result_tool,
"result_args": result_args,
"result_pending": result_pending,
"status": status,
"source": source,
"reasoning": reasoning,
"elapsed": elapsed,
"error": None,
}
def run_oneshot_phase(pipeline, filter_category: Optional[str] = None) -> list:
"""Execute la phase one-shot pour tous les tests.
Args:
pipeline: Pipeline initialise
filter_category: Si donne, limite aux tests de cette categorie (ex: "FEDORA")
Returns:
list[dict]: resultats bruts par test
"""
results = []
for cat, desc, query, expected_tool, mandatory_args, optional_args in TESTS:
top_cat = cat.split("/")[0]
if filter_category and top_cat.upper() != filter_category.upper():
results.append(None) # Placeholder pour aligner avec dry-run
continue
r = run_oneshot_single(pipeline, query, expected_tool, mandatory_args, optional_args)
r.update({"cat": cat, "desc": desc, "query": query, "expected_tool": expected_tool})
results.append(r)
return results
# --- Rapport ---
def _status_color(status: str, no_color: bool = False) -> str:
if no_color:
return ""
colors = {
PASS: "\033[32m",
PARTIAL: "\033[33m",
FAIL: "\033[31m",
RULE_MISS: "\033[34m",
ERROR: "\033[35m",
}
return colors.get(status, "")
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
CYAN = "\033[36m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW= "\033[33m"
def print_report(dry_results: list, one_results: list,
fails_only: bool = False, no_color: bool = False):
"""Affiche le rapport combine DRY-RUN + ONE-SHOT."""
def c(color): return "" if no_color else color
total = len(TESTS)
# --- Compteurs ---
def count(results, status):
return sum(1 for r in results if r and r.get("status") == status)
dry_pass = count(dry_results, PASS)
dry_partial = count(dry_results, PARTIAL)
dry_fail = count(dry_results, FAIL)
dry_miss = count(dry_results, RULE_MISS)
one_pass = count(one_results, PASS)
one_partial = count(one_results, PARTIAL)
one_fail = count(one_results, FAIL)
one_miss = count(one_results, RULE_MISS)
one_error = count(one_results, ERROR)
one_skip = sum(1 for r in one_results if r is None)
# Coherence: dry et oneshot ont le meme statut
coherent = sum(
1 for d, o in zip(dry_results, one_results)
if d and o and d["status"] == o["status"]
)
compared = sum(1 for d, o in zip(dry_results, one_results) if d and o)
sep = "=" * 70
print(f"\n{c(BOLD)}{sep}{c(RESET)}")
print(f" {c(BOLD)}TEST CAMPAIGN ONE-SHOT - Lyra RAG v2{c(RESET)}")
print(f" {total} tests | Phase 1: dry-run (rules) | Phase 2: one-shot (pipeline)")
print(f"{c(BOLD)}{sep}{c(RESET)}")
# --- Details par test ---
prev_top = None
for i, (dry, one) in enumerate(zip(dry_results, one_results)):
if dry is None:
continue
cat = dry["cat"]
top_cat = cat.split("/")[0]
query = dry["query"]
ds = dry["status"]
os_ = one["status"] if one else "SKIP"
# Filtrage
if fails_only and ds == PASS and os_ == PASS:
continue
# Separateur categorie
if top_cat != prev_top:
print(f"\n {c(BOLD)}{c(CYAN)}--- {top_cat} ---{c(RESET)}")
prev_top = top_cat
# Ligne de resultat
num = f"{i+1:>3}/{total}"
dry_label = f"{c(_status_color(ds, no_color))}{ds:<9}{c(RESET)}"
one_label = f"{c(_status_color(os_, no_color))}{os_:<9}{c(RESET)}" if one else f"{c(DIM)}SKIP {c(RESET)}"
diverge_marker = ""
if one and ds != os_:
diverge_marker = f" {c(YELLOW)}[!]{c(RESET)}"
short_q = query[:50].ljust(50) if len(query) <= 50 else query[:47] + "..."
print(f" [{num}] DRY:{dry_label} ONE:{one_label} {c(DIM)}{short_q}{c(RESET)}{diverge_marker}")
# Details en cas d'echec ou divergence
if ds != PASS or (one and os_ != PASS):
if one and one.get("result_tool") and one["result_tool"] != dry["result_tool"]:
print(f" DRY tool : {dry['result_tool'] or 'none'}")
print(f" ONE tool : {one['result_tool'] or 'none'} [{one.get('source','?')}]")
elif ds != PASS:
print(f" tool : {dry['result_tool'] or 'none'}")
if one and os_ == ERROR:
print(f" {c(RED)}ERROR: {one.get('error','')}{c(RESET)}")
if one and os_ == FAIL and one.get("reasoning"):
if "IntentClassifier" in one["reasoning"]:
print(f" {c(RED)}{one['reasoning']}{c(RESET)}")
# --- Scores ---
print(f"\n{c(BOLD)}{sep}{c(RESET)}")
print(f" {c(BOLD)}SCORE FINAL{c(RESET)}")
print(f"{c(BOLD)}{sep}{c(RESET)}")
print(f" {'':20} {'DRY-RUN':>12} {'ONE-SHOT':>12}")
print(f" {'-'*46}")
def pct(n, d): return f"{100*n//d}%" if d > 0 else "N/A"
one_done = total - one_skip
print(f" {c(GREEN)}PASS{c(RESET)}{'':16} {dry_pass:>6}/{total} {pct(dry_pass,total):>4} {one_pass:>6}/{one_done} {pct(one_pass,one_done):>4}")
print(f" {c(YELLOW)}PARTIAL{c(RESET)}{'':13} {dry_partial:>6}/{total} {pct(dry_partial,total):>4} {one_partial:>6}/{one_done} {pct(one_partial,one_done):>4}")
print(f" RULE_MISS{'':11} {dry_miss:>6}/{total} {pct(dry_miss,total):>4} {one_miss:>6}/{one_done} {pct(one_miss,one_done):>4}")
print(f" {c(RED)}FAIL{c(RESET)}{'':16} {dry_fail:>6}/{total} {pct(dry_fail,total):>4} {one_fail:>6}/{one_done} {pct(one_fail,one_done):>4}")
if one_error:
print(f" {c(RED)}ERROR (exception){c(RESET)} {' ':>10} {one_error:>6}/{one_done}")
if one_skip:
print(f" {c(DIM)}SKIP (filtre){c(RESET)}{'':8} {' ':>10} {one_skip:>6}/{total}")
if compared > 0:
print(f"\n Coherence DRY==ONE : {coherent}/{compared} ({100*coherent//compared}%)")
# --- Divergences ---
divergences = [
(i, dry_results[i], one_results[i])
for i in range(total)
if dry_results[i] and one_results[i]
and dry_results[i]["status"] != one_results[i]["status"]
]
if divergences:
print(f"\n{c(BOLD)}{sep}{c(RESET)}")
print(f" {c(BOLD)}{c(YELLOW)}DIVERGENCES (dry != oneshot){c(RESET)}")
print(f"{c(BOLD)}{sep}{c(RESET)}")
for i, dry, one in divergences:
print(f"\n [{i+1:>3}] {dry['cat']} | {dry['desc']}")
print(f" Query : {dry['query']}")
print(f" Attendu : {dry['expected_tool']}")
print(f" DRY : {c(_status_color(dry['status'], no_color))}{dry['status']}{c(RESET)} -> {dry['result_tool'] or 'none'}")
one_src = f" [{one.get('source','?')} {one.get('elapsed',0):.1f}s]" if one.get('elapsed') else ""
print(f" ONE-SHOT : {c(_status_color(one['status'], no_color))}{one['status']}{c(RESET)} -> {one['result_tool'] or 'none'}{one_src}")
if one.get("reasoning"):
print(f" Raison : {one['reasoning'][:80]}")
print(f"{c(BOLD)}{sep}{c(RESET)}\n")
# --- Main ---
def main():
parser = argparse.ArgumentParser(
description="Campagne de tests ONE-SHOT Lyra RAG v2",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Exemples:\n"
" python3 tests/test_campaign_oneshot.py\n"
" python3 tests/test_campaign_oneshot.py --category FEDORA\n"
" python3 tests/test_campaign_oneshot.py --fails-only\n"
" python3 tests/test_campaign_oneshot.py --dry-only\n"
)
)
parser.add_argument("--category", "-c", default=None,
help="Filtrer par categorie (ex: FEDORA, HUE, TV, CATT, DENON, BACKUP)")
parser.add_argument("--fails-only", action="store_true",
help="N'afficher que les tests avec au moins un echec")
parser.add_argument("--dry-only", action="store_true",
help="Phase 1 uniquement (sans Ollama, rapide)")
parser.add_argument("--oneshot-only", action="store_true",
help="Phase 2 uniquement (sans re-afficher dry-run)")
parser.add_argument("--no-color", action="store_true", help="Desactiver les couleurs ANSI")
parser.add_argument("--ephaistos", default=None,
help="Modele EPHAISTOS a mesurer (ex: qwen2.5-coder:7b). "
"Ecrit une configuration derivee, config.yaml n'est jamais modifie.")
parser.add_argument("--lyra", default=None, help="Modele LYRA a mesurer")
parser.add_argument("--json", action="store_true",
help="Publie le resultat dans benchmarks/results/")
args = parser.parse_args()
# --- Phase 1 : DRY-RUN ---
print(f"\n{BOLD}Phase 1 : DRY-RUN (rules uniquement){RESET}", flush=True)
t0 = time.perf_counter()
dry_results = run_dryrun_phase()
dry_elapsed = time.perf_counter() - t0
dry_pass = sum(1 for r in dry_results if r["status"] == PASS)
print(f" Termine en {dry_elapsed:.2f}s | {dry_pass}/{len(TESTS)} PASS", flush=True)
if args.dry_only:
# Creer des placeholders pour le rapport
one_results = [None] * len(TESTS)
print_report(dry_results, one_results, fails_only=args.fails_only, no_color=args.no_color)
sys.exit(0 if dry_pass == len(TESTS) else 1)
# --- Phase 2 : ONE-SHOT ---
print(f"\n{BOLD}Phase 2 : ONE-SHOT (pipeline complet){RESET}", flush=True)
print(" Initialisation Ollama + RAG + MCP...", end="", flush=True)
t_init = time.perf_counter()
chemin_config = None
if args.ephaistos or args.lyra:
import tempfile
import yaml as _yaml
cfg = _yaml.safe_load((ROOT / "config.yaml").read_text()) or {}
cfg.setdefault("models", {})
if args.ephaistos:
cfg["models"].setdefault("ephaistos", {})["name"] = args.ephaistos
if args.lyra:
cfg["models"].setdefault("lyra", {})["name"] = args.lyra
tmp = tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False)
_yaml.safe_dump(cfg, tmp, allow_unicode=True)
tmp.close()
chemin_config = tmp.name
print(f" Configuration derivee : ephaistos={args.ephaistos or '(inchange)'} "
f"lyra={args.lyra or '(inchange)'}", flush=True)
try:
pipeline = init_pipeline(chemin_config)
except Exception as exc:
print(f"\n {RED}ERREUR init pipeline: {exc}{RESET}")
print(" Verifiez qu'Ollama est actif et que les modeles sont disponibles.")
sys.exit(2)
init_elapsed = time.perf_counter() - t_init
print(f" OK ({init_elapsed:.1f}s)", flush=True)
# Filtrer les tests si --category
total_filtered = sum(
1 for cat, *_ in TESTS
if not args.category or cat.split("/")[0].upper() == args.category.upper()
)
print(f" Execution de {total_filtered} tests...", flush=True)
one_results = []
t_run = time.perf_counter()
for idx, (cat, desc, query, expected_tool, mandatory_args, optional_args) in enumerate(TESTS):
top_cat = cat.split("/")[0]
if args.category and top_cat.upper() != args.category.upper():
one_results.append(None)
continue
# Progression inline
print(f" [{idx+1:>3}/{len(TESTS)}] {query[:55]:<55}", end="\r", flush=True)
r = run_oneshot_single(pipeline, query, expected_tool, mandatory_args, optional_args)
r.update({"cat": cat, "desc": desc, "query": query, "expected_tool": expected_tool})
one_results.append(r)
run_elapsed = time.perf_counter() - t_run
one_pass = sum(1 for r in one_results if r and r.get("status") == PASS)
one_done = sum(1 for r in one_results if r is not None)
print(f" Termine en {run_elapsed:.1f}s | {one_pass}/{one_done} PASS{' ' * 20}", flush=True)
# --- Rapport ---
print_report(dry_results, one_results, fails_only=args.fails_only, no_color=args.no_color)
# Code retour: 0 si aucun FAIL ni RULE_MISS. PARTIAL est tolere :
# l'outil detecte est correct mais le workflow est interactif (ex: choix
# COW du clone) — ce n'est pas un echec de detection.
one_bad = sum(1 for r in one_results
if r and r.get("status") in (FAIL, RULE_MISS))
if args.json:
from collections import Counter
sys.path.insert(0, str(ROOT / "scripts"))
from bench_common import ecrire_resultat, modeles # noqa: E402
actifs = modeles()
if args.ephaistos:
actifs["ephaistos"] = args.ephaistos
if args.lyra:
actifs["lyra"] = args.lyra
statuts = Counter(r["status"] for r in one_results if r)
par_cat = {}
for r in one_results:
if not r:
continue
racine = r["cat"].split("/")[0]
par_cat.setdefault(racine, Counter())[r["status"]] += 1
slug = (args.ephaistos or actifs.get("ephaistos", "defaut")).replace(":", "-")
chemin = ecrire_resultat("campagne", {
"banc": "one-shot (pipeline complet, avec LLM)",
"cas": one_done,
"duree_s": round(run_elapsed, 1),
"statuts": dict(statuts),
"taux_pass": round(one_pass / one_done, 4) if one_done else 0.0,
"par_categorie": {c: dict(v) for c, v in sorted(par_cat.items())},
"modeles_mesures": actifs,
}, suffixe=slug)
print(f"Resume publie : {chemin}")
all_ok = (dry_pass == len(TESTS)) and one_bad == 0
sys.exit(0 if all_ok else 1)
if __name__ == "__main__":
main()