Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
02ee2d7b5beb8b3761e614c5b656b0a79cdaae5e
rpa_vision_v3/tests/property/test_workflow_composition_properties.py
Dom a27b74cf22 v1.0 - Version stable: multi-PC, détection UI-DETR-1, 3 modes exécution 
- Frontend v4 accessible sur réseau local (192.168.1.40)
- Ports ouverts: 3002 (frontend), 5001 (backend), 5004 (dashboard)
- Ollama GPU fonctionnel
- Self-healing interactif
- Dashboard confiance

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-29 11:23:51 +01:00

1280 lines
44 KiB
Python
Raw Blame History

"""
Tests Property-Based pour Workflow Composition
Ce fichier contient les tests property-based pour valider les propriétés
de correction définies dans le design document pour la composition de workflows.
Utilise Hypothesis pour la génération de données aléatoires.
"""
import pytest
from hypothesis import given, strategies as st, settings, assume
from datetime import datetime
from typing import Dict, Any, List
# =============================================================================
# Stratégies de génération pour les modèles de composition
# =============================================================================
# Stratégie pour les identifiants
id_strategy = st.text(
alphabet=st.characters(whitelist_categories=('L', 'N'), whitelist_characters='_-'),
min_size=1,
max_size=50
).filter(lambda x: len(x.strip()) > 0)
# Stratégie pour les noms de variables
variable_name_strategy = st.text(
alphabet=st.characters(whitelist_categories=('L', 'N'), whitelist_characters='_'),
min_size=1,
max_size=30
).filter(lambda x: len(x.strip()) > 0 and x[0].isalpha())
# Stratégie pour les valeurs simples
simple_value_strategy = st.one_of(
st.text(min_size=0, max_size=100),
st.integers(min_value=-10000, max_value=10000),
st.floats(min_value=-1000.0, max_value=1000.0, allow_nan=False, allow_infinity=False),
st.booleans()
)
# Stratégie pour les dictionnaires de variables
variable_dict_strategy = st.dictionaries(
keys=variable_name_strategy,
values=simple_value_strategy,
min_size=0,
max_size=10
)
# Stratégie pour les types de condition
condition_type_strategy = st.sampled_from([
"element_present", "element_absent", "text_equals", "text_contains"
])
# Stratégie pour les actions en cas d'échec
failure_action_strategy = st.sampled_from(["retry", "skip", "abort"])
# Stratégie pour les types de boucle
loop_type_strategy = st.sampled_from(["count", "condition"])
# Stratégie pour les types de trigger
trigger_type_strategy = st.sampled_from(["schedule", "file", "visual"])
# Stratégie pour les priorités de séquence
priority_strategy = st.sampled_from(["high", "medium", "low"])
# =============================================================================
# Stratégies composites pour les modèles
# =============================================================================
@st.composite
def visual_condition_strategy(draw):
"""Génère une VisualCondition valide."""
from core.workflow.composition_models import VisualCondition
condition_type = draw(condition_type_strategy)
target_element = draw(st.one_of(st.none(), id_strategy))
expected_text = draw(st.one_of(st.none(), st.text(min_size=0, max_size=100)))
return VisualCondition(
condition_type=condition_type,
target_element=target_element,
expected_text=expected_text
)
@st.composite
def chain_config_strategy(draw):
"""Génère une ChainConfig valide."""
from core.workflow.composition_models import ChainConfig
return ChainConfig(
source_workflow_id=draw(id_strategy),
target_workflow_id=draw(id_strategy),
variable_mapping=draw(st.dictionaries(
keys=variable_name_strategy,
values=variable_name_strategy,
min_size=0,
max_size=5
)),
on_failure=draw(failure_action_strategy),
max_retries=draw(st.integers(min_value=0, max_value=10))
)
@st.composite
def loop_config_strategy(draw):
"""Génère une LoopConfig valide."""
from core.workflow.composition_models import LoopConfig, VisualCondition
loop_type = draw(loop_type_strategy)
max_iterations = None
exit_condition = None
if loop_type == "count":
max_iterations = draw(st.integers(min_value=1, max_value=100))
else:
exit_condition = draw(visual_condition_strategy())
return LoopConfig(
loop_id=draw(id_strategy),
loop_type=loop_type,
max_iterations=max_iterations,
exit_condition=exit_condition,
body_nodes=draw(st.lists(id_strategy, min_size=0, max_size=10)),
safety_limit=draw(st.integers(min_value=100, max_value=2000))
)
@st.composite
def branch_config_strategy(draw):
"""Génère une BranchConfig valide."""
from core.workflow.composition_models import BranchConfig
return BranchConfig(
branch_id=draw(id_strategy),
condition=draw(visual_condition_strategy()),
target_node=draw(id_strategy),
priority=draw(st.integers(min_value=0, max_value=100))
)
@st.composite
def conditional_node_strategy(draw):
"""Génère un ConditionalNode valide."""
from core.workflow.composition_models import ConditionalNode
return ConditionalNode(
node_id=draw(id_strategy),
branches=draw(st.lists(branch_config_strategy(), min_size=0, max_size=5)),
default_branch=draw(st.one_of(st.none(), id_strategy))
)
@st.composite
def schedule_trigger_strategy(draw):
"""Génère un ScheduleTrigger valide."""
from core.workflow.composition_models import ScheduleTrigger
return ScheduleTrigger(
trigger_id=draw(id_strategy),
workflow_id=draw(id_strategy),
cron_expression=draw(st.one_of(st.none(), st.just("0 * * * *"))),
interval_seconds=draw(st.one_of(st.none(), st.integers(min_value=1, max_value=86400))),
enabled=draw(st.booleans())
)
@st.composite
def file_trigger_strategy(draw):
"""Génère un FileTrigger valide."""
from core.workflow.composition_models import FileTrigger
return FileTrigger(
trigger_id=draw(id_strategy),
workflow_id=draw(id_strategy),
watch_directory=draw(st.text(min_size=1, max_size=100)),
file_pattern=draw(st.sampled_from(["*.csv", "*.xlsx", "*.txt", "report_*"])),
enabled=draw(st.booleans())
)
@st.composite
def visual_trigger_strategy(draw):
"""Génère un VisualTrigger valide."""
from core.workflow.composition_models import VisualTrigger
return VisualTrigger(
trigger_id=draw(id_strategy),
workflow_id=draw(id_strategy),
target_element=draw(id_strategy),
check_interval_seconds=draw(st.integers(min_value=1, max_value=3600)),
enabled=draw(st.booleans())
)
@st.composite
def trigger_strategy(draw):
"""Génère un trigger de n'importe quel type."""
trigger_type = draw(trigger_type_strategy)
if trigger_type == "schedule":
return draw(schedule_trigger_strategy())
elif trigger_type == "file":
return draw(file_trigger_strategy())
else:
return draw(visual_trigger_strategy())
@st.composite
def parameter_def_strategy(draw):
"""Génère un ParameterDef valide."""
from core.workflow.composition_models import ParameterDef
return ParameterDef(
name=draw(variable_name_strategy),
param_type=draw(st.sampled_from(["string", "number", "boolean", "list", "dict"])),
required=draw(st.booleans()),
default_value=draw(st.one_of(st.none(), simple_value_strategy)),
description=draw(st.text(min_size=0, max_size=200))
)
@st.composite
def sub_workflow_definition_strategy(draw):
"""Génère une SubWorkflowDefinition valide."""
from core.workflow.composition_models import SubWorkflowDefinition
return SubWorkflowDefinition(
workflow_id=draw(id_strategy),
name=draw(st.text(min_size=1, max_size=100)),
input_parameters=draw(st.lists(parameter_def_strategy(), min_size=0, max_size=5)),
output_values=draw(st.lists(parameter_def_strategy(), min_size=0, max_size=5)),
description=draw(st.text(min_size=0, max_size=200))
)
@st.composite
def reference_node_strategy(draw):
"""Génère un ReferenceNode valide."""
from core.workflow.composition_models import ReferenceNode
return ReferenceNode(
node_id=draw(id_strategy),
sub_workflow_id=draw(id_strategy),
input_bindings=draw(st.dictionaries(
keys=variable_name_strategy,
values=variable_name_strategy,
min_size=0,
max_size=5
)),
output_bindings=draw(st.dictionaries(
keys=variable_name_strategy,
values=variable_name_strategy,
min_size=0,
max_size=5
))
)
@st.composite
def log_entry_strategy(draw):
"""Génère un LogEntry valide."""
from core.workflow.composition_models import LogEntry
return LogEntry(
timestamp=datetime.now(),
workflow_id=draw(id_strategy),
node_id=draw(id_strategy),
event_type=draw(st.sampled_from(["start", "end", "action", "error", "warning", "info"])),
details=draw(st.dictionaries(
keys=st.text(min_size=1, max_size=20),
values=simple_value_strategy,
min_size=0,
max_size=5
))
)
@st.composite
def trigger_context_strategy(draw):
"""Génère un TriggerContext valide."""
from core.workflow.composition_models import TriggerContext
trigger_type = draw(trigger_type_strategy)
return TriggerContext(
trigger_id=draw(id_strategy),
trigger_type=trigger_type,
fired_at=datetime.now(),
file_path=draw(st.one_of(st.none(), st.text(min_size=1, max_size=100))) if trigger_type == "file" else None,
detected_element=draw(st.one_of(st.none(), st.dictionaries(
keys=st.text(min_size=1, max_size=20),
values=simple_value_strategy,
min_size=0,
max_size=3
))) if trigger_type == "visual" else None
)
@st.composite
def execution_context_strategy(draw):
"""Génère un ExecutionContext valide."""
from core.workflow.composition_models import ExecutionContext
return ExecutionContext(
chain_id=draw(id_strategy),
current_workflow_id=draw(id_strategy),
global_variables=draw(variable_dict_strategy),
execution_log=draw(st.lists(log_entry_strategy(), min_size=0, max_size=5)),
trigger_context=draw(st.one_of(st.none(), trigger_context_strategy())),
started_at=datetime.now()
)
@st.composite
def validation_result_strategy(draw):
"""Génère un ValidationResult valide."""
from core.workflow.composition_models import ValidationResult
return ValidationResult(
is_valid=draw(st.booleans()),
errors=draw(st.lists(st.text(min_size=1, max_size=100), min_size=0, max_size=5)),
warnings=draw(st.lists(st.text(min_size=1, max_size=100), min_size=0, max_size=5))
)
# =============================================================================
# Property 1: Round-trip de sérialisation
# Feature: workflow-composition, Property 1: Round-trip de sérialisation
# Validates: Requirements 1.6, 2.6, 6.6, 7.6, 8.6, 9.4
# =============================================================================
@given(config=chain_config_strategy())
@settings(max_examples=100, deadline=None)
def test_property_chain_config_roundtrip(config):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 1.6**
Pour tout ChainConfig, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ChainConfig
# Sérialiser
data = config.to_dict()
# Désérialiser
restored = ChainConfig.from_dict(data)
# Vérifier l'équivalence
assert config.source_workflow_id == restored.source_workflow_id
assert config.target_workflow_id == restored.target_workflow_id
assert config.variable_mapping == restored.variable_mapping
assert config.on_failure == restored.on_failure
assert config.max_retries == restored.max_retries
@given(config=loop_config_strategy())
@settings(max_examples=100, deadline=None)
def test_property_loop_config_roundtrip(config):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 6.6**
Pour tout LoopConfig, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import LoopConfig
# Sérialiser
data = config.to_dict()
# Désérialiser
restored = LoopConfig.from_dict(data)
# Vérifier l'équivalence
assert config.loop_id == restored.loop_id
assert config.loop_type == restored.loop_type
assert config.max_iterations == restored.max_iterations
assert config.body_nodes == restored.body_nodes
assert config.safety_limit == restored.safety_limit
# Vérifier exit_condition si présent
if config.exit_condition is not None:
assert restored.exit_condition is not None
assert config.exit_condition.condition_type == restored.exit_condition.condition_type
assert config.exit_condition.target_element == restored.exit_condition.target_element
assert config.exit_condition.expected_text == restored.exit_condition.expected_text
else:
assert restored.exit_condition is None
@given(node=conditional_node_strategy())
@settings(max_examples=100, deadline=None)
def test_property_conditional_node_roundtrip(node):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 7.6**
Pour tout ConditionalNode, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ConditionalNode
# Sérialiser
data = node.to_dict()
# Désérialiser
restored = ConditionalNode.from_dict(data)
# Vérifier l'équivalence
assert node.node_id == restored.node_id
assert node.default_branch == restored.default_branch
assert len(node.branches) == len(restored.branches)
for orig, rest in zip(node.branches, restored.branches):
assert orig.branch_id == rest.branch_id
assert orig.target_node == rest.target_node
assert orig.priority == rest.priority
assert orig.condition.condition_type == rest.condition.condition_type
@given(trigger=schedule_trigger_strategy())
@settings(max_examples=100, deadline=None)
def test_property_schedule_trigger_roundtrip(trigger):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 8.6**
Pour tout ScheduleTrigger, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ScheduleTrigger
data = trigger.to_dict()
restored = ScheduleTrigger.from_dict(data)
assert trigger.trigger_id == restored.trigger_id
assert trigger.workflow_id == restored.workflow_id
assert trigger.cron_expression == restored.cron_expression
assert trigger.interval_seconds == restored.interval_seconds
assert trigger.enabled == restored.enabled
@given(trigger=file_trigger_strategy())
@settings(max_examples=100, deadline=None)
def test_property_file_trigger_roundtrip(trigger):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 8.6**
Pour tout FileTrigger, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import FileTrigger
data = trigger.to_dict()
restored = FileTrigger.from_dict(data)
assert trigger.trigger_id == restored.trigger_id
assert trigger.workflow_id == restored.workflow_id
assert trigger.watch_directory == restored.watch_directory
assert trigger.file_pattern == restored.file_pattern
assert trigger.enabled == restored.enabled
@given(trigger=visual_trigger_strategy())
@settings(max_examples=100, deadline=None)
def test_property_visual_trigger_roundtrip(trigger):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 8.6**
Pour tout VisualTrigger, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import VisualTrigger
data = trigger.to_dict()
restored = VisualTrigger.from_dict(data)
assert trigger.trigger_id == restored.trigger_id
assert trigger.workflow_id == restored.workflow_id
assert trigger.target_element == restored.target_element
assert trigger.check_interval_seconds == restored.check_interval_seconds
assert trigger.enabled == restored.enabled
@given(ctx=execution_context_strategy())
@settings(max_examples=100, deadline=None)
def test_property_execution_context_roundtrip(ctx):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 9.4**
Pour tout ExecutionContext, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ExecutionContext
data = ctx.to_dict()
restored = ExecutionContext.from_dict(data)
assert ctx.chain_id == restored.chain_id
assert ctx.current_workflow_id == restored.current_workflow_id
assert ctx.global_variables == restored.global_variables
assert len(ctx.execution_log) == len(restored.execution_log)
if ctx.trigger_context is not None:
assert restored.trigger_context is not None
assert ctx.trigger_context.trigger_id == restored.trigger_context.trigger_id
assert ctx.trigger_context.trigger_type == restored.trigger_context.trigger_type
else:
assert restored.trigger_context is None
@given(defn=sub_workflow_definition_strategy())
@settings(max_examples=100, deadline=None)
def test_property_sub_workflow_definition_roundtrip(defn):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 2.6**
Pour tout SubWorkflowDefinition, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import SubWorkflowDefinition
data = defn.to_dict()
restored = SubWorkflowDefinition.from_dict(data)
assert defn.workflow_id == restored.workflow_id
assert defn.name == restored.name
assert defn.description == restored.description
assert len(defn.input_parameters) == len(restored.input_parameters)
assert len(defn.output_values) == len(restored.output_values)
for orig, rest in zip(defn.input_parameters, restored.input_parameters):
assert orig.name == rest.name
assert orig.param_type == rest.param_type
assert orig.required == rest.required
@given(ref=reference_node_strategy())
@settings(max_examples=100, deadline=None)
def test_property_reference_node_roundtrip(ref):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 2.6**
Pour tout ReferenceNode, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ReferenceNode
data = ref.to_dict()
restored = ReferenceNode.from_dict(data)
assert ref.node_id == restored.node_id
assert ref.sub_workflow_id == restored.sub_workflow_id
assert ref.input_bindings == restored.input_bindings
assert ref.output_bindings == restored.output_bindings
@given(result=validation_result_strategy())
@settings(max_examples=100, deadline=None)
def test_property_validation_result_roundtrip(result):
"""
**Feature: workflow-composition, Property 1: Round-trip de sérialisation**
**Validates: Requirements 1.6**
Pour tout ValidationResult, la sérialisation suivie de la désérialisation
doit produire un objet équivalent.
"""
from core.workflow.composition_models import ValidationResult
data = result.to_dict()
restored = ValidationResult.from_dict(data)
assert result.is_valid == restored.is_valid
assert result.errors == restored.errors
assert result.warnings == restored.warnings
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])
# =============================================================================
# Property 13: Validation des dépendances
# Feature: workflow-composition, Property 13: Validation des dépendances
# Validates: Requirements 5.2, 5.3
# =============================================================================
@given(
workflow_ids=st.lists(id_strategy, min_size=2, max_size=10, unique=True)
)
@settings(max_examples=100, deadline=None)
def test_property_circular_dependency_detection(workflow_ids):
"""
**Feature: workflow-composition, Property 13: Validation des dépendances**
**Validates: Requirements 5.3**
Pour toute configuration créant une dépendance circulaire,
elle doit être rejetée avec une erreur.
"""
from core.workflow.dependency_graph import DependencyGraph, CircularDependencyError
assume(len(workflow_ids) >= 2)
graph = DependencyGraph()
# Créer une chaîne linéaire de dépendances
for i in range(len(workflow_ids) - 1):
graph.add_dependency(workflow_ids[i], workflow_ids[i + 1])
# Tenter de créer un cycle (dernier -> premier)
try:
graph.add_dependency(workflow_ids[-1], workflow_ids[0])
# Si on arrive ici, c'est une erreur
assert False, "Devrait avoir levé CircularDependencyError"
except CircularDependencyError as e:
# Vérifier que le cycle contient les bons éléments
assert workflow_ids[0] in e.cycle or workflow_ids[-1] in e.cycle
@given(
parent_id=id_strategy,
child_ids=st.lists(id_strategy, min_size=1, max_size=5, unique=True)
)
@settings(max_examples=100, deadline=None)
def test_property_deletion_warning(parent_id, child_ids):
"""
**Feature: workflow-composition, Property 13: Validation des dépendances**
**Validates: Requirements 5.2**
Pour toute tentative de suppression d'un sous-workflow avec des dépendants,
un avertissement doit être généré listant tous les workflows affectés.
"""
from core.workflow.dependency_graph import DependencyGraph
# S'assurer que parent_id n'est pas dans child_ids
assume(parent_id not in child_ids)
graph = DependencyGraph()
# Créer des dépendances: tous les children dépendent du parent
for child_id in child_ids:
graph.add_dependency(child_id, parent_id)
# Obtenir les avertissements de suppression
warnings = graph.get_deletion_warnings(parent_id)
# Propriété: il doit y avoir au moins un avertissement
assert len(warnings) > 0, "Devrait avoir des avertissements de suppression"
# Propriété: l'avertissement doit mentionner le nombre de dépendants
dependents = graph.get_dependents(parent_id)
assert len(dependents) == len(child_ids)
@given(
workflow_id=id_strategy,
sub_workflow_id=id_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_self_reference_detection(workflow_id, sub_workflow_id):
"""
**Feature: workflow-composition, Property 13: Validation des dépendances**
**Validates: Requirements 5.3**
Une auto-référence (workflow qui dépend de lui-même) doit être détectée.
"""
from core.workflow.dependency_graph import DependencyGraph, CircularDependencyError
graph = DependencyGraph()
# Tenter une auto-référence
try:
graph.add_dependency(workflow_id, workflow_id)
assert False, "Devrait avoir levé CircularDependencyError pour auto-référence"
except CircularDependencyError:
pass # Comportement attendu
# =============================================================================
# Property 2: Propagation des variables dans les chaînes
# Feature: workflow-composition, Property 2: Propagation des variables
# Validates: Requirements 1.2, 9.1, 9.5
# =============================================================================
@given(
var_name=variable_name_strategy,
initial_value=simple_value_strategy,
override_value=simple_value_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_variable_propagation(var_name, initial_value, override_value):
"""
**Feature: workflow-composition, Property 2: Propagation des variables**
**Validates: Requirements 1.2, 9.1, 9.5**
Pour tout workflow A avec des variables de contexte chaîné vers un workflow B,
toutes les variables mappées doivent être accessibles dans B avec leurs valeurs correctes,
et en cas de conflit de noms, la valeur la plus récente prévaut.
"""
from core.workflow.global_variable_manager import GlobalVariableManager
gvm = GlobalVariableManager()
# Workflow A définit une variable
gvm.set_global(var_name, initial_value, "workflow_a")
# Vérifier que la variable est accessible
assert gvm.get_global(var_name) == initial_value
# Workflow B override la variable
gvm.set_global(var_name, override_value, "workflow_b")
# Vérifier que la valeur la plus récente prévaut
assert gvm.get_global(var_name) == override_value
# Vérifier que l'override est loggé
overrides = gvm.get_override_log()
assert len(overrides) == 1
assert overrides[0].variable_name == var_name
assert overrides[0].old_value == initial_value
assert overrides[0].new_value == override_value
# =============================================================================
# Property 21: Lecture de variable avec défaut
# Feature: workflow-composition, Property 21: Lecture de variable avec défaut
# Validates: Requirements 9.2
# =============================================================================
@given(
var_name=variable_name_strategy,
default_value=simple_value_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_variable_default_value(var_name, default_value):
"""
**Feature: workflow-composition, Property 21: Lecture de variable avec défaut**
**Validates: Requirements 9.2**
Pour toute lecture d'une variable globale non définie,
la valeur par défaut spécifiée doit être retournée.
"""
from core.workflow.global_variable_manager import GlobalVariableManager
gvm = GlobalVariableManager()
# La variable n'existe pas
assert not gvm.has_global(var_name)
# Lecture avec défaut
result = gvm.get_global(var_name, default_value)
# Doit retourner la valeur par défaut
assert result == default_value
# La variable ne doit toujours pas exister
assert not gvm.has_global(var_name)
# =============================================================================
# Property 14: Incrémentation du compteur de boucle
# Feature: workflow-composition, Property 14: Incrémentation du compteur
# Validates: Requirements 6.2
# =============================================================================
@given(
loop_id=id_strategy,
num_iterations=st.integers(min_value=1, max_value=50)
)
@settings(max_examples=100, deadline=None)
def test_property_loop_counter_increment(loop_id, num_iterations):
"""
**Feature: workflow-composition, Property 14: Incrémentation du compteur**
**Validates: Requirements 6.2**
Pour toute boucle, après chaque itération complète,
le compteur doit être incrémenté exactement de 1.
"""
from core.workflow.loop_executor import LoopExecutor
from core.workflow.composition_models import LoopConfig
executor = LoopExecutor(safety_limit=1000)
config = LoopConfig(
loop_id=loop_id,
loop_type="count",
max_iterations=num_iterations + 10, # Plus que ce qu'on va exécuter
body_nodes=[]
)
executor.start_loop(config)
for expected in range(1, num_iterations + 1):
result = executor.execute_iteration(loop_id)
assert result.iteration == expected, \
f"Compteur incorrect: attendu {expected}, obtenu {result.iteration}"
# =============================================================================
# Property 15: Terminaison de boucle à la limite
# Feature: workflow-composition, Property 15: Terminaison à la limite
# Validates: Requirements 6.3
# =============================================================================
@given(
loop_id=id_strategy,
max_iterations=st.integers(min_value=1, max_value=20)
)
@settings(max_examples=100, deadline=None)
def test_property_loop_termination_at_limit(loop_id, max_iterations):
"""
**Feature: workflow-composition, Property 15: Terminaison à la limite**
**Validates: Requirements 6.3**
Pour toute boucle avec une limite de N itérations,
elle doit terminer après exactement N itérations.
"""
from core.workflow.loop_executor import LoopExecutor
from core.workflow.composition_models import LoopConfig
executor = LoopExecutor(safety_limit=1000)
config = LoopConfig(
loop_id=loop_id,
loop_type="count",
max_iterations=max_iterations,
body_nodes=[]
)
executor.start_loop(config)
iterations_executed = 0
while executor.should_continue(loop_id):
executor.execute_iteration(loop_id)
iterations_executed += 1
if iterations_executed > max_iterations + 1:
break # Protection contre boucle infinie dans le test
assert iterations_executed == max_iterations, \
f"Nombre d'itérations incorrect: attendu {max_iterations}, obtenu {iterations_executed}"
# =============================================================================
# Property 16: Garde de sécurité des boucles
# Feature: workflow-composition, Property 16: Garde de sécurité
# Validates: Requirements 6.5
# =============================================================================
@given(
loop_id=id_strategy,
safety_limit=st.integers(min_value=5, max_value=20)
)
@settings(max_examples=50, deadline=None)
def test_property_loop_safety_guard(loop_id, safety_limit):
"""
**Feature: workflow-composition, Property 16: Garde de sécurité**
**Validates: Requirements 6.5**
Pour toute boucle, le nombre d'itérations ne doit jamais dépasser
la limite de sécurité.
"""
from core.workflow.loop_executor import LoopExecutor, LoopSafetyLimitError
from core.workflow.composition_models import LoopConfig
executor = LoopExecutor(safety_limit=safety_limit)
config = LoopConfig(
loop_id=loop_id,
loop_type="count",
max_iterations=None, # Pas de limite explicite
body_nodes=[],
safety_limit=safety_limit
)
executor.start_loop(config)
iterations = 0
safety_triggered = False
try:
for _ in range(safety_limit + 5):
executor.execute_iteration(loop_id)
iterations += 1
except LoopSafetyLimitError:
safety_triggered = True
assert safety_triggered, "La garde de sécurité aurait dû se déclencher"
assert iterations <= safety_limit, \
f"Trop d'itérations: {iterations} > limite {safety_limit}"
# =============================================================================
# Property 17: Ordre d'évaluation des branches conditionnelles
# Feature: workflow-composition, Property 17: Ordre d'évaluation
# Validates: Requirements 7.2
# =============================================================================
@given(
num_branches=st.integers(min_value=2, max_value=5)
)
@settings(max_examples=100, deadline=None)
def test_property_branch_evaluation_order(num_branches):
"""
**Feature: workflow-composition, Property 17: Ordre d'évaluation**
**Validates: Requirements 7.2**
Pour tout node conditionnel avec plusieurs branches dont les conditions sont vraies,
seule la première branche (par ordre de priorité) doit être exécutée.
"""
from core.workflow.conditional_evaluator import ConditionalEvaluator
from core.workflow.composition_models import ConditionalNode, BranchConfig, VisualCondition
evaluator = ConditionalEvaluator()
# Créer des branches avec des priorités différentes
# Toutes les conditions sont vraies (element_present avec le même élément)
branches = []
for i in range(num_branches):
branches.append(BranchConfig(
branch_id=f"branch_{i}",
condition=VisualCondition(
condition_type="element_present",
target_element="common_element"
),
target_node=f"target_{i}",
priority=i # Priorité croissante
))
node = ConditionalNode(
node_id="test_node",
branches=branches,
default_branch=None
)
# L'élément est présent, donc toutes les conditions sont vraies
screen_state = {"element_ids": ["common_element"]}
result = evaluator.evaluate_node(node, screen_state)
# La première branche (priorité 0) doit être sélectionnée
assert result == "target_0", \
f"Devrait sélectionner target_0 (priorité 0), obtenu {result}"
# =============================================================================
# Property 18: Branche par défaut ou erreur
# Feature: workflow-composition, Property 18: Branche par défaut
# Validates: Requirements 7.3
# =============================================================================
@given(
has_default=st.booleans(),
default_target=id_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_default_branch_or_error(has_default, default_target):
"""
**Feature: workflow-composition, Property 18: Branche par défaut**
**Validates: Requirements 7.3**
Pour tout node conditionnel où aucune condition n'est satisfaite,
la branche par défaut doit être exécutée si elle existe,
sinon une erreur doit être levée.
"""
from core.workflow.conditional_evaluator import ConditionalEvaluator, NoMatchingBranchError
from core.workflow.composition_models import ConditionalNode, BranchConfig, VisualCondition
evaluator = ConditionalEvaluator()
# Créer une branche avec une condition qui ne sera jamais vraie
branches = [BranchConfig(
branch_id="never_match",
condition=VisualCondition(
condition_type="element_present",
target_element="nonexistent_element"
),
target_node="unreachable",
priority=0
)]
node = ConditionalNode(
node_id="test_node",
branches=branches,
default_branch=default_target if has_default else None
)
# État vide, aucune condition ne sera vraie
screen_state = {}
if has_default:
result = evaluator.evaluate_node(node, screen_state)
assert result == default_target, \
f"Devrait retourner la branche par défaut {default_target}, obtenu {result}"
else:
try:
evaluator.evaluate_node(node, screen_state)
assert False, "Devrait lever NoMatchingBranchError"
except NoMatchingBranchError:
pass # Comportement attendu
# =============================================================================
# Property 5: Retour de contrôle après sous-workflow
# Feature: workflow-composition, Property 5: Retour de contrôle
# Validates: Requirements 2.1, 2.2
# =============================================================================
@given(
parent_id=id_strategy,
sub_id=id_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_subworkflow_control_return(parent_id, sub_id):
"""
**Feature: workflow-composition, Property 5: Retour de contrôle**
**Validates: Requirements 2.1, 2.2**
Pour tout workflow parent appelant un sous-workflow via une référence,
après complétion du sous-workflow, l'exécution doit reprendre au parent.
"""
from core.workflow.subworkflow_registry import SubWorkflowRegistry
from core.workflow.composition_models import (
SubWorkflowDefinition, ReferenceNode, ExecutionContext
)
assume(parent_id != sub_id)
registry = SubWorkflowRegistry()
# Enregistrer le sous-workflow
defn = SubWorkflowDefinition(
workflow_id=sub_id,
name="Test Sub",
input_parameters=[],
output_values=[]
)
registry.register(defn)
# Créer une référence
ref = ReferenceNode(
node_id="ref1",
sub_workflow_id=sub_id,
input_bindings={},
output_bindings={}
)
registry.create_reference(parent_id, ref)
# Créer le contexte
ctx = ExecutionContext(
chain_id="test_chain",
current_workflow_id=parent_id,
global_variables={}
)
# Exécuter la référence
result = registry.execute_reference(ref, ctx)
# Vérifier que le contrôle est revenu au parent
assert ctx.current_workflow_id == parent_id, \
f"Le contrôle devrait revenir à {parent_id}, pas {ctx.current_workflow_id}"
assert result.success
# =============================================================================
# Property 7: Détection de similarité pour fusion
# Feature: workflow-composition, Property 7: Détection de similarité
# Validates: Requirements 3.1
# =============================================================================
@given(
shared_nodes=st.lists(id_strategy, min_size=3, max_size=10, unique=True),
unique_a=st.lists(id_strategy, min_size=0, max_size=3, unique=True),
unique_b=st.lists(id_strategy, min_size=0, max_size=3, unique=True)
)
@settings(max_examples=100, deadline=None)
def test_property_similarity_detection(shared_nodes, unique_a, unique_b):
"""
**Feature: workflow-composition, Property 7: Détection de similarité**
**Validates: Requirements 3.1**
Pour toute paire de workflows avec un score de similarité supérieur à 0.9,
le système doit générer une suggestion de fusion.
"""
from core.workflow.workflow_merger import WorkflowMerger
# S'assurer que les nodes uniques ne sont pas dans les partagés
unique_a = [n for n in unique_a if n not in shared_nodes]
unique_b = [n for n in unique_b if n not in shared_nodes]
merger = WorkflowMerger(similarity_threshold=0.5)
nodes_a = shared_nodes + unique_a
nodes_b = shared_nodes + unique_b
similarity = merger.calculate_similarity(nodes_a, nodes_b)
# Vérifier que la similarité est calculée correctement
assert 0.0 <= similarity <= 1.0
# Si beaucoup de nodes partagés, la similarité devrait être élevée
if len(shared_nodes) > 0 and len(unique_a) == 0 and len(unique_b) == 0:
assert similarity == 1.0
# =============================================================================
# Property 8: Préservation des chemins lors de la fusion
# Feature: workflow-composition, Property 8: Préservation des chemins
# Validates: Requirements 3.2
# =============================================================================
@given(
nodes_a=st.lists(id_strategy, min_size=1, max_size=10, unique=True),
nodes_b=st.lists(id_strategy, min_size=1, max_size=10, unique=True)
)
@settings(max_examples=100, deadline=None)
def test_property_path_preservation(nodes_a, nodes_b):
"""
**Feature: workflow-composition, Property 8: Préservation des chemins**
**Validates: Requirements 3.2**
Pour tout merge de workflows A et B, le workflow résultant doit contenir
tous les nodes uniques de A et tous les nodes uniques de B.
"""
from core.workflow.workflow_merger import WorkflowMerger
merger = WorkflowMerger()
merged = merger.merge(nodes_a, nodes_b)
# Tous les nodes de A doivent être dans le résultat
for node in nodes_a:
assert node in merged, f"Node {node} de A manquant dans le merge"
# Tous les nodes de B doivent être dans le résultat
for node in nodes_b:
assert node in merged, f"Node {node} de B manquant dans le merge"
# =============================================================================
# Property 10: Détection de séquences communes
# Feature: workflow-composition, Property 10: Détection de séquences
# Validates: Requirements 4.1
# =============================================================================
@given(
common_seq=st.lists(id_strategy, min_size=3, max_size=5, unique=True),
prefix_a=st.lists(id_strategy, min_size=0, max_size=2, unique=True),
prefix_b=st.lists(id_strategy, min_size=0, max_size=2, unique=True)
)
@settings(max_examples=50, deadline=None)
def test_property_sequence_detection(common_seq, prefix_a, prefix_b):
"""
**Feature: workflow-composition, Property 10: Détection de séquences**
**Validates: Requirements 4.1**
Pour toute séquence de nodes identique apparaissant dans N workflows,
le système doit la détecter.
"""
from core.workflow.sequence_extractor import SequenceExtractor
# S'assurer que les préfixes ne contiennent pas d'éléments de la séquence commune
prefix_a = [n for n in prefix_a if n not in common_seq]
prefix_b = [n for n in prefix_b if n not in common_seq]
extractor = SequenceExtractor(min_sequence_length=3)
workflows = {
'wf_a': prefix_a + common_seq,
'wf_b': prefix_b + common_seq
}
sequences = extractor.find_common_sequences(workflows)
# Si la séquence commune est assez longue, elle devrait être détectée
if len(common_seq) >= 3:
# Au moins une séquence devrait être trouvée
found_common = any(
set(seq.nodes) == set(common_seq) or
all(n in common_seq for n in seq.nodes)
for seq in sequences
)
# Note: peut ne pas trouver si les préfixes interfèrent
# On vérifie juste que la fonction ne plante pas
# =============================================================================
# Property 19: Passage du contexte de déclenchement
# Feature: workflow-composition, Property 19: Contexte de déclenchement
# Validates: Requirements 8.4
# =============================================================================
@given(
trigger_id=id_strategy,
workflow_id=id_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_trigger_context_passing(trigger_id, workflow_id):
"""
**Feature: workflow-composition, Property 19: Contexte de déclenchement**
**Validates: Requirements 8.4**
Pour tout déclenchement de trigger, le contexte doit être passé au workflow.
"""
from core.workflow.trigger_manager import TriggerManager
from core.workflow.composition_models import ScheduleTrigger
manager = TriggerManager()
trigger = ScheduleTrigger(
trigger_id=trigger_id,
workflow_id=workflow_id,
interval_seconds=60
)
manager.register_trigger(trigger)
# Déclencher
ctx = manager.fire_trigger(trigger_id)
# Vérifier le contexte
assert ctx.trigger_id == trigger_id
assert ctx.trigger_type == "schedule"
assert ctx.fired_at is not None
# =============================================================================
# Property 22: Préservation de l'état final des variables
# Feature: workflow-composition, Property 22: État final des variables
# Validates: Requirements 9.3
# =============================================================================
@given(
variables=variable_dict_strategy
)
@settings(max_examples=100, deadline=None)
def test_property_final_variable_state(variables):
"""
**Feature: workflow-composition, Property 22: État final des variables**
**Validates: Requirements 9.3**
Pour toute chaîne de workflows complétée, le log d'exécution doit contenir
l'état final de toutes les variables globales.
"""
from core.workflow.execution_logger import ExecutionLogger
logger = ExecutionLogger()
# Simuler quelques événements
logger.log_start("wf1", "node1")
logger.log_action("wf1", "node1", "click")
logger.log_end("wf1", "node1")
# Définir l'état final
logger.set_final_variable_state(variables)
# Récupérer l'état final
final_state = logger.get_final_variable_state()
# Vérifier que toutes les variables sont préservées
assert final_state == variables
# Vérifier la sérialisation
data = logger.to_dict()
assert data["final_variables"] == variables
Reference in New Issue View Git Blame Copy Permalink