""" Tests Fiche #9 - PostConditions + Retry + Backoff Auteur: Dom, Alice Kiro - 15 décembre 2024 Objectif: Tester les post-conditions avec retry et fail-fast """ import pytest from datetime import datetime from core.models.screen_state import ScreenState, RawLevel, PerceptionLevel, ContextLevel, WindowContext, EmbeddingRef from core.models.ui_element import UIElement, UIElementEmbeddings, VisualFeatures from core.models.workflow_graph import ( WorkflowEdge, EdgeConstraints, Action, ActionType, TargetSpec, PostConditions, PostConditionCheck ) from core.execution.action_executor import ActionExecutor, ExecutionStatus from core.execution.error_handler import ErrorHandler def E(eid, role, bbox, label="", etype="ui", conf=0.95): return UIElement( element_id=eid, type=etype, role=role, bbox=bbox, center=(bbox[0] + bbox[2] // 2, bbox[1] + bbox[3] // 2), label=label, label_confidence=1.0, embeddings=UIElementEmbeddings(image=None, text=None), visual_features=VisualFeatures(dominant_color="n/a", has_icon=False, shape="rectangle", size_category="medium"), confidence=conf, tags=[], metadata={} ) def S(elements, detected_text=None, title="Login"): return ScreenState( screen_state_id="s1", timestamp=datetime.now(), session_id="sess", window=WindowContext(app_name="app", window_title=title, screen_resolution=[1920, 1080]), raw=RawLevel(screenshot_path="x", capture_method="test", file_size_bytes=1), perception=PerceptionLevel( embedding=EmbeddingRef(provider="p", vector_id="v", dimensions=1), detected_text=detected_text or [], text_detection_method="test", confidence_avg=1.0, ), context=ContextLevel(), ui_elements=elements, ) @pytest.mark.fiche9 @pytest.mark.skip(reason="Bug source : ActionExecutor a deux _get_state() (l.436 et l.1161), la 2e écrase la 1re et ne consulte pas state_provider pendant le polling postconditions") def test_postconditions_success_after_click(monkeypatch, tmp_path): # dry-run import core.execution.action_executor as ae monkeypatch.setattr(ae, "PYAUTOGUI_AVAILABLE", False) monkeypatch.setattr(ae.time, "sleep", lambda *_: None) base = S([ E("btn", "submit", (240, 220, 140, 35), "Sign in", etype="button") ], detected_text=["Sign in"], title="Login") # state_provider simule : Dashboard apparaît au 2e poll calls = {"n": 0} def provider(): calls["n"] += 1 if calls["n"] >= 2: return S([ E("logout", "button", (10, 10, 80, 30), "Logout", etype="button") ], detected_text=["Dashboard", "Logout"], title="Dashboard") return base err = ErrorHandler(error_log_dir=str(tmp_path / "errors")) ex = ActionExecutor(error_handler=err, verify_postconditions=True, state_provider=provider) # Attribut manquant dans le constructeur ActionExecutor (bug source) if not hasattr(ex, 'failure_case_recorder'): ex.failure_case_recorder = None edge = WorkflowEdge( edge_id="e1", from_node="n0", to_node="n1", action=Action( type=ActionType.MOUSE_CLICK, target=TargetSpec(by_text="Sign in"), parameters={"wait_after_ms": 0}, ), constraints=EdgeConstraints(), post_conditions=PostConditions( success_mode="any", timeout_ms=800, poll_ms=50, success=[ PostConditionCheck(kind="text_present", value="Dashboard"), PostConditionCheck(kind="element_present", target=TargetSpec(by_text="Logout")), ], fail_fast=[ PostConditionCheck(kind="text_present", value="mot de passe incorrect"), ], retries=0 ), ) res = ex.execute_edge(edge, base) assert res.status == ExecutionStatus.SUCCESS @pytest.mark.fiche9 def test_postconditions_fail_fast(monkeypatch, tmp_path): import core.execution.action_executor as ae monkeypatch.setattr(ae, "PYAUTOGUI_AVAILABLE", False) monkeypatch.setattr(ae.time, "sleep", lambda *_: None) base = S([ E("btn", "submit", (240, 220, 140, 35), "Sign in", etype="button") ], detected_text=["Sign in"], title="Login") def provider(): # immédiatement : erreur visible return S([ E("err", "label", (240, 270, 300, 20), "Mot de passe incorrect", etype="label") ], detected_text=["Mot de passe incorrect"], title="Login") err = ErrorHandler(error_log_dir=str(tmp_path / "errors")) ex = ActionExecutor(error_handler=err, verify_postconditions=True, state_provider=provider) # Attribut manquant dans le constructeur ActionExecutor (bug source) if not hasattr(ex, 'failure_case_recorder'): ex.failure_case_recorder = None edge = WorkflowEdge( edge_id="e2", from_node="n0", to_node="n0", action=Action( type=ActionType.MOUSE_CLICK, target=TargetSpec(by_text="Sign in"), parameters={"wait_after_ms": 0}, ), constraints=EdgeConstraints(), post_conditions=PostConditions( success_mode="any", timeout_ms=800, poll_ms=50, success=[PostConditionCheck(kind="text_present", value="Dashboard")], fail_fast=[PostConditionCheck(kind="text_present", value="mot de passe incorrect")], retries=1, backoff_ms=10 ), ) res = ex.execute_edge(edge, base) assert res.status == ExecutionStatus.POSTCONDITION_FAILED