tests/integration/test_vm_clone_running.py
Rama main · réplica de lectura del repositorio original.
"""
Test d'integration pour le workflow de clone avec VM running.
Verifie que la detection de VM en cours d'execution fonctionne
dans le workflow multi-tour (clarification step-by-step).
"""
from unittest.mock import Mock
import pytest
from lyra.core.config import RAGConfig
from lyra.core.pipeline import Pipeline
from lyra.hestia.executor import ExecutionResult
from lyra.models.ephaistos import EphaistosAnalysis
from lyra.models.intent_classifier import ClassificationResult, Intent
from lyra.models.lyra_voice import LyraResponse
@pytest.fixture
def pipeline_with_mocks():
"""Pipeline avec tous les composants mockes."""
config = RAGConfig()
config.llm = {"base_url": "http://localhost:11434", "timeout": 120}
config.mcp = {"servers": {}}
pipeline = Pipeline(config)
# Mock les composants
# Fournir un doc minimal pour que la pipeline passe a EPHAISTOS
mock_spec = Mock()
mock_spec.document = "vm_clone(source_vm: string, new_vm_name: string) - Clone une VM KVM"
mock_spec.metadata = {"name": "fedora.vm_clone", "server": "fedora"}
mock_spec.score = 0.9
pipeline._retriever = Mock()
pipeline._retriever.retrieve.return_value = [mock_spec]
pipeline._model_manager = Mock()
pipeline._ephaistos = Mock()
# _compact_spec est appele par le pipeline (filtrer_missing_args) : un Mock casse re.split
pipeline._ephaistos._compact_spec.side_effect = lambda spec: spec
pipeline._lyra = Mock()
pipeline._hestia = Mock()
pipeline._intent_classifier = Mock()
# Session memory reelle pour tester le multi-tour
from lyra.rag.session_memory import SessionMemory
pipeline._sessions["default"] = SessionMemory(max_turns=10)
pipeline._initialized = True
# Creer WorkflowContext avec les mocks
return pipeline
def test_vm_clone_running_detection_multiturn(pipeline_with_mocks):
"""
Test le workflow complet de clone avec detection de VM running
dans le cas multi-tour (clarification step-by-step).
Scenario:
1. User: "clone preprod01" (avec typo)
2. Pipeline: detecte typo, demande confirmation source
3. User: "preprod-01" (correction)
4. Pipeline: demande destination
5. User: "clone4test"
6. Pipeline: DOIT detecter que preprod-01 est running et proposer 3 options
"""
pipeline = pipeline_with_mocks
# ============================================================
# Configuration des mocks
# ============================================================
# Intent classifier: toujours "demande"
pipeline._intent_classifier.classify.return_value = ClassificationResult(
intent=Intent.DEMANDE,
confidence=0.95,
raw_response="Action MCP"
)
# Mock LYRA pour acknowledgements et clarifications
pipeline._lyra.generate_acknowledgement.return_value = None
# Mock HESTIA pour vm_status
def mock_execute(tool_name, arguments=None, **kwargs):
"""Mock de l'executeur Hestia."""
if arguments is None:
arguments = {}
# get_all_vms ou vm_status sans vm_name
if "get_all_vms" in tool_name or (tool_name == "fedora.vm_status" and not arguments.get("vm_name")):
content = "preprod-01 running 192.168.122.245 22\npreprod-02 stopped - -\n"
return ExecutionResult(success=True, content=content)
# vm_status avec vm_name=preprod-01 (RUNNING)
elif tool_name == "fedora.vm_status" and arguments.get("vm_name") == "preprod-01":
content = (
" ========================================\n"
" VM: preprod-01\n"
" ========================================\n\n"
" Status: en cours d'execution\n"
" IP: 192.168.122.245\n"
" SSH: (22)\n"
)
return ExecutionResult(success=True, content=content)
# vm_status avec vm_name=preprod-02 (STOPPED)
elif tool_name == "fedora.vm_status" and arguments.get("vm_name") == "preprod-02":
content = (
" ========================================\n"
" VM: preprod-02\n"
" ========================================\n\n"
" Status: Arretee\n"
" IP: -\n"
)
return ExecutionResult(success=True, content=content)
return ExecutionResult(success=False, content="Unknown tool")
pipeline._hestia.execute = mock_execute
# ============================================================
# TOUR 1: "clone preprod01" (avec typo)
# ============================================================
# EPHAISTOS analyse Tour 1
pipeline._ephaistos.analyze_with_retry.return_value = EphaistosAnalysis(
tool="fedora.vm_clone",
arguments={"source_vm": "preprod01"}, # avec typo
missing_args=["new_vm_name"],
confidence=0.8,
reasoning="Clone VM",
raw_response=""
)
result1 = pipeline.process("clone preprod01")
# Verification Tour 1
assert "preprod-01" in result1.response.lower(), "Tour 1: devrait suggerer preprod-01"
assert result1.pending_args, "Tour 1: devrait avoir pending args"
print("\n✅ TOUR 1: Typo detectee")
print(f" Response: {result1.response[:100]}...")
# ============================================================
# TOUR 2: "preprod-01" (correction source)
# ============================================================
# EPHAISTOS extract_missing_args Tour 2
# L'utilisateur confirme "preprod-01", on demande maintenant new_vm_name
pipeline._ephaistos.extract_missing_args.return_value = EphaistosAnalysis(
tool="fedora.vm_clone",
arguments={"source_vm": "preprod-01"}, # source corrigee
missing_args=["new_vm_name"], # encore besoin du nom destination
confidence=0.9,
reasoning="Source confirmee",
raw_response=""
)
# LYRA demande le nom destination
pipeline._lyra.ask_clarification.return_value = LyraResponse(
text="Quel nom veux-tu pour le clone ?",
mentions_ephaistos=False,
mentions_hestia=False
)
result2 = pipeline.process("preprod-01")
# Verification Tour 2
assert "nom" in result2.response.lower() or "clone" in result2.response.lower(), \
"Tour 2: devrait demander le nom destination"
print("\n✅ TOUR 2: Demande nom destination")
print(f" Response: {result2.response[:100]}...")
# ============================================================
# TOUR 3: "clone4test" (nom destination) - LE TEST CRITIQUE
# ============================================================
# EPHAISTOS extract_missing_args Tour 3: TOUS les arguments sont presents
pipeline._ephaistos.extract_missing_args.return_value = EphaistosAnalysis(
tool="fedora.vm_clone",
arguments={
"source_vm": "preprod-01",
"new_vm_name": "clone4test"
},
missing_args=[], # PLUS D'ARGUMENTS MANQUANTS
confidence=0.95,
reasoning="Arguments complets",
raw_response=""
)
result3 = pipeline.process("clone4test")
# ============================================================
# VERIFICATION FINALE - LE TEST CRITIQUE
# ============================================================
print(f"\n{'='*70}")
print("TOUR 3: Detection VM running (TEST CRITIQUE)")
print(f"{'='*70}")
print(f"\nResponse complete:\n{result3.response}")
print(f"\n{'='*70}")
# Le Tour 3 DOIT detecter que preprod-01 est running et proposer 3 options
expected_phrases = [
"en cours", # "en cours d'execution"
"arrêter", # "arreter la VM"
"option", # "Options:"
"1", # Option 1
"2", # Option 2
"3", # Option 3
"annuler" # Option 3: Annuler
]
found = []
missing = []
for phrase in expected_phrases:
if phrase.lower() in result3.response.lower():
found.append(phrase)
else:
missing.append(phrase)
# Assertions
assert "en cours" in result3.response.lower(), \
f"CRITICAL: La detection de VM running ne fonctionne pas!\n" \
f"Response: {result3.response[:300]}\n" \
f"Phrases manquantes: {missing}"
assert "option" in result3.response.lower(), \
f"CRITICAL: Le menu d'options n'apparait pas!\n" \
f"Response: {result3.response[:300]}"
assert "1" in result3.response and "2" in result3.response and "3" in result3.response, \
f"CRITICAL: Les 3 options ne sont pas toutes presentes!\n" \
f"Response: {result3.response[:300]}"
print("\n✅ ✅ ✅ TEST REUSSI ✅ ✅ ✅")
print("\nLa detection de VM running fonctionne correctement!")
print("Le menu avec 3 options s'affiche comme attendu.")
print(f"\nPhrases trouvees: {found}")
def test_vm_clone_stopped_vm_no_menu(pipeline_with_mocks):
"""
Test que le menu d'options n'apparait PAS si la VM est arretee.
"""
pipeline = pipeline_with_mocks
# Intent classifier
pipeline._intent_classifier.classify.return_value = ClassificationResult(
intent=Intent.DEMANDE,
confidence=0.95,
raw_response="Action MCP"
)
pipeline._lyra.generate_acknowledgement.return_value = None
# Mock HESTIA: preprod-02 est ARRETEE
def mock_execute(tool_name, arguments=None, **kwargs):
if arguments is None:
arguments = {}
if "get_all_vms" in tool_name or (tool_name == "fedora.vm_status" and not arguments.get("vm_name")):
content = "preprod-01 running 192.168.122.245 22\npreprod-02 stopped - -\n"
return ExecutionResult(success=True, content=content)
elif tool_name == "fedora.vm_status" and arguments.get("vm_name") == "preprod-02":
content = (
" ========================================\n"
" VM: preprod-02\n"
" ========================================\n\n"
" Status: Arretee\n"
" IP: -\n"
)
return ExecutionResult(success=True, content=content)
return ExecutionResult(success=False, content="Unknown")
pipeline._hestia.execute = mock_execute
# Analyse directe avec tous les arguments
pipeline._ephaistos.analyze_with_retry.return_value = EphaistosAnalysis(
tool="fedora.vm_clone",
arguments={
"source_vm": "preprod-02",
"new_vm_name": "test-clone"
},
missing_args=[],
confidence=0.95,
reasoning="Clone VM arretee",
raw_response=""
)
result = pipeline.process("clone preprod-02 en test-clone")
# Verification: PAS de menu d'options car VM arretee
assert "option" not in result.response.lower(), \
"Ne devrait PAS afficher le menu d'options si la VM est arretee"
assert "en cours d'exécution" not in result.response.lower(), \
"Ne devrait PAS indiquer que la VM est running si elle est arretee"
print("\n✅ TEST REUSSI: Pas de menu pour VM arretee")
print(f" Response: {result.response[:150]}...")