""" Tests unitaires pour les composants de workflow Teste: - VariableManager : Gestion des variables - SemanticMatcher : Matching sémantique """ import pytest import json import tempfile from pathlib import Path from core.workflow import ( VariableManager, VariableDefinition, SemanticMatcher, WorkflowMatch, create_variable_manager_from_workflow ) # ============================================================================= # Tests VariableManager # ============================================================================= class TestVariableManager: """Tests pour VariableManager.""" def test_define_variable(self): """Test définition de variable.""" vm = VariableManager() vm.define_variable("client", "Nom du client", required=True) definitions = vm.get_definitions() assert "client" in definitions assert definitions["client"].required is True def test_set_and_get_variable(self): """Test set/get de variable.""" vm = VariableManager() vm.set_variable("name", "John") assert vm.get_variable("name") == "John" assert vm.get_variable("unknown") is None assert vm.get_variable("unknown", "default") == "default" def test_substitute_simple(self): """Test substitution simple.""" vm = VariableManager() vm.set_variable("client", "Acme") result = vm.substitute("Facturer {{client}}") assert result == "Facturer Acme" def test_substitute_with_default(self): """Test substitution avec valeur par défaut.""" vm = VariableManager() result = vm.substitute("Montant: {{montant|0}} euros") assert result == "Montant: 0 euros" def test_substitute_missing_variable(self): """Test substitution avec variable manquante.""" vm = VariableManager() result = vm.substitute("Client: {{client}}") # Variable non définie reste telle quelle assert "{{client}}" in result def test_substitute_dict(self): """Test substitution dans un dictionnaire.""" vm = VariableManager() vm.set_variable("client", "Acme") vm.set_variable("montant", "1000") data = { "name": "Facture {{client}}", "amount": "{{montant}}", "nested": { "description": "Pour {{client}}" } } result = vm.substitute_dict(data) assert result["name"] == "Facture Acme" assert result["amount"] == "1000" assert result["nested"]["description"] == "Pour Acme" def test_extract_variables(self): """Test extraction de variables.""" vm = VariableManager() text = "Facturer {{client}} pour {{montant}} euros" variables = vm.extract_variables(text) assert "client" in variables assert "montant" in variables assert len(variables) == 2 def test_validation_required(self): """Test validation des variables requises.""" vm = VariableManager() vm.define_variable("client", required=True) vm.define_variable("optional", required=False) # Sans valeur pour client errors = vm.validate() assert len(errors) == 1 assert "client" in errors[0] # Avec valeur pour client vm.set_variable("client", "Acme") errors = vm.validate() assert len(errors) == 0 def test_validation_with_default(self): """Test validation avec valeur par défaut.""" vm = VariableManager() vm.define_variable("montant", required=True, default_value="0") # La valeur par défaut satisfait la validation errors = vm.validate() assert len(errors) == 0 def test_type_conversion(self): """Test conversion de type.""" vm = VariableManager() vm.define_variable("count", var_type="integer") vm.define_variable("active", var_type="boolean") vm.set_variable("count", "42") vm.set_variable("active", "true") assert vm.get_variable("count") == 42 assert vm.get_variable("active") is True def test_serialization(self): """Test sérialisation/désérialisation.""" vm = VariableManager() vm.define_variable("client", "Nom", required=True) vm.set_variable("client", "Acme") # Sérialiser data = vm.to_dict() # Désérialiser vm2 = VariableManager.from_dict(data) assert vm2.get_variable("client") == "Acme" assert "client" in vm2.get_definitions() # ============================================================================= # Tests SemanticMatcher # ============================================================================= class TestSemanticMatcher: """Tests pour SemanticMatcher.""" @pytest.fixture def temp_workflows_dir(self): """Créer un répertoire temporaire avec des workflows de test.""" with tempfile.TemporaryDirectory() as tmpdir: workflows_dir = Path(tmpdir) # Créer workflow de facturation facturation = { "name": "Facturation Client", "description": "Créer une facture pour un client", "tags": ["facturer", "facture", "client", "invoice"], "param_patterns": [ "(?:client|customer)\\s+(?P[A-Za-z0-9_\\-]+)" ] } with open(workflows_dir / "facturation.json", "w") as f: json.dump(facturation, f) # Créer workflow d'export export = { "name": "Export Rapport", "description": "Exporter un rapport", "tags": ["export", "rapport", "pdf", "excel"], "param_patterns": [ "(?:format|en)\\s+(?Ppdf|excel)" ] } with open(workflows_dir / "export.json", "w") as f: json.dump(export, f) yield workflows_dir def test_load_workflows(self, temp_workflows_dir): """Test chargement des workflows.""" matcher = SemanticMatcher(str(temp_workflows_dir)) workflows = matcher.get_all_workflows() assert len(workflows) == 2 def test_find_workflow_exact_match(self, temp_workflows_dir): """Test matching exact.""" matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("Facturation Client") assert match is not None assert match.workflow_name == "Facturation Client" assert match.confidence > 0.5 def test_find_workflow_by_tag(self, temp_workflows_dir): """Test matching par tag.""" matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("facturer quelque chose") assert match is not None assert "Facturation" in match.workflow_name def test_find_workflow_by_keywords(self, temp_workflows_dir): """Test matching par mots-clés.""" matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("créer une facture") assert match is not None assert "Facturation" in match.workflow_name def test_extract_params(self, temp_workflows_dir): """Test extraction de paramètres.""" matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("facturer client Acme") assert match is not None assert "client" in match.extracted_params assert match.extracted_params["client"].lower() == "acme" def test_extract_range_params(self, temp_workflows_dir): """Test extraction de paramètres de plage.""" matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("facturer de A à Z") assert match is not None assert "start" in match.extracted_params assert "end" in match.extracted_params def test_find_multiple_workflows(self, temp_workflows_dir): """Test recherche de plusieurs workflows.""" matcher = SemanticMatcher(str(temp_workflows_dir)) matches = matcher.find_workflows("rapport", limit=5) assert len(matches) >= 1 assert any("Export" in m.workflow_name for m in matches) def test_min_confidence_filter(self, temp_workflows_dir): """Test filtre de confiance minimale.""" matcher = SemanticMatcher(str(temp_workflows_dir)) # Avec confiance élevée, moins de résultats matches_high = matcher.find_workflows("xyz random", min_confidence=0.8) matches_low = matcher.find_workflows("xyz random", min_confidence=0.1) assert len(matches_high) <= len(matches_low) def test_suggest_commands(self, temp_workflows_dir): """Test suggestions de commandes.""" matcher = SemanticMatcher(str(temp_workflows_dir)) suggestions = matcher.suggest_commands("Fact") assert len(suggestions) > 0 assert any("Facturation" in s for s in suggestions) def test_get_workflow_help(self, temp_workflows_dir): """Test aide pour un workflow.""" matcher = SemanticMatcher(str(temp_workflows_dir)) help_text = matcher.get_workflow_help("facturation") assert "Facturation Client" in help_text assert "Tags" in help_text # ============================================================================= # Tests d'intégration # ============================================================================= class TestWorkflowIntegration: """Tests d'intégration VariableManager + SemanticMatcher.""" @pytest.fixture def temp_workflows_dir(self): """Créer un répertoire temporaire avec un workflow paramétré.""" with tempfile.TemporaryDirectory() as tmpdir: workflows_dir = Path(tmpdir) workflow = { "name": "Facturation", "description": "Facturer {{client}} pour {{montant}} euros", "tags": ["facturer"], "variables": [ {"name": "client", "required": True}, {"name": "montant", "required": False, "default_value": "0"} ], "edges": [ { "action": { "type": "text_input", "parameters": {"text": "{{client}}"} } } ] } with open(workflows_dir / "facturation.json", "w") as f: json.dump(workflow, f) yield workflows_dir def test_full_workflow_execution(self, temp_workflows_dir): """Test exécution complète avec variables.""" # 1. Trouver le workflow matcher = SemanticMatcher(str(temp_workflows_dir)) match = matcher.find_workflow("facturer client Acme") assert match is not None # 2. Charger le workflow with open(match.workflow_path) as f: workflow_data = json.load(f) # 3. Créer le VariableManager vm = create_variable_manager_from_workflow(workflow_data) # 4. Injecter les paramètres extraits vm.set_variables(match.extracted_params) # 5. Substituer les variables result = vm.substitute_dict(workflow_data) # Vérifier la substitution (le matcher normalise en minuscules) assert "acme" in result["description"].lower() assert result["edges"][0]["action"]["parameters"]["text"].lower() == "acme" if __name__ == "__main__": pytest.main([__file__, "-v"])