""" Base Simulation Framework - Abstract base class for GUI automation tests. This module provides the foundation for all simulation-based tests in the Manual Slop test suite. Simulations act as external "puppeteers" that drive the GUI through the ApiHookClient HTTP interface. Architecture: - BaseSimulation: Abstract base class with setup/teardown lifecycle - WorkflowSimulator: High-level workflow operations (project setup, file mgmt) - ApiHookClient: Low-level HTTP client for Hook API communication Typical Usage: class MySimulation(BaseSimulation): def run(self) -> None: self.client.set_value('mma_epic_input', 'My epic description') self.client.click('btn_mma_plan_epic') # Poll for completion... status = self.client.get_mma_status() assert status['mma_status'] == 'done' if __name__ == '__main__': run_sim(MySimulation) Lifecycle: 1. setup() - Connects to GUI, resets session, scaffolds temp project 2. run() - Implemented by subclass with simulation logic 3. teardown() - Cleanup (optional file retention for debugging) Prerequisites: - GUI must be running with --enable-test-hooks flag - HookServer must be listening on http://127.0.0.1:8999 Thread Safety: - Simulations are designed to run in the main thread - ApiHookClient handles its own connection pooling See Also: - simulation/workflow_sim.py for WorkflowSimulator - tests/conftest.py for live_gui pytest fixture - docs/guide_simulations.md for full simulation documentation """ import sys import os import time from typing import Any, Optional from api_hook_client import ApiHookClient from simulation.workflow_sim import WorkflowSimulator # Ensure project root and src/ are in path project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) sys.path.append(project_root) sys.path.append(os.path.join(project_root, "src")) class BaseSimulation: def __init__(self, client: ApiHookClient = None) -> None: if client is None: self.client = ApiHookClient() else: self.client = client self.sim = WorkflowSimulator(self.client) self.project_path = None def setup(self, project_name: str = "SimProject") -> None: """ [C: simulation/sim_execution.py:ExecutionSimulation.setup, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_extended_sims.py:test_context_sim_live, tests/test_extended_sims.py:test_execution_sim_live, tests/test_extended_sims.py:test_tools_sim_live, tests/test_sim_base.py:test_base_simulation_setup] """ print("\n[BaseSim] Connecting to GUI...") if not self.client.wait_for_server(timeout=5): raise RuntimeError("Could not connect to GUI. Ensure it is running with --enable-test-hooks") self.client.clear_events() self.client.set_value("auto_add_history", True) # Wait for propagation _start = time.time() while time.time() - _start < 5.0: if self.client.get_value("auto_add_history") is True: break time.sleep(0.1) print("[BaseSim] Resetting session...") self.client.click("btn_reset") time.sleep(2.0) git_dir = os.path.abspath(".") self.project_path = os.path.abspath(f"tests/artifacts/temp_{project_name.lower()}.toml") if os.path.exists(self.project_path): os.remove(self.project_path) print(f"[BaseSim] Scaffolding Project: {project_name}") self.sim.setup_new_project(project_name, git_dir, self.project_path) # Standard test settings self.client.set_value("current_provider", "gemini") self.client.set_value("current_model", "gemini-2.5-flash-lite") time.sleep(1.5) def teardown(self) -> None: """ [C: tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_extended_sims.py:test_context_sim_live, tests/test_extended_sims.py:test_execution_sim_live, tests/test_extended_sims.py:test_tools_sim_live] """ if self.project_path and os.path.exists(self.project_path): # We keep it for debugging if it failed, but usually we'd clean up # os.remove(self.project_path) pass print("[BaseSim] Teardown complete.") def get_value(self, tag: str) -> Any: """ [C: simulation/sim_context.py:ContextSimulation.run, simulation/sim_execution.py:ExecutionSimulation.run, simulation/workflow_sim.py:WorkflowSimulator.run_discussion_turn_async, simulation/workflow_sim.py:WorkflowSimulator.wait_for_ai_response, tests/smoke_status_hook.py:test_status_hook, tests/smoke_status_hook.py:wait_for_value, tests/test_auto_switch_sim.py:test_auto_switch_sim, tests/test_deepseek_infra.py:test_gui_provider_list_via_hooks, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_gui2_parity.py:test_gui2_click_hook_works, tests/test_gui2_parity.py:test_gui2_set_value_hook_works, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_rag_visual_sim.py:test_rag_full_lifecycle_sim, tests/test_rag_visual_sim.py:test_rag_settings_persistence_sim, tests/test_selectable_ui.py:test_selectable_label_stability, tests/test_system_prompt_sim.py:test_system_prompt_sim, tests/test_undo_redo_sim.py:test_undo_redo_context_mutation, tests/test_undo_redo_sim.py:test_undo_redo_discussion_mutation, tests/test_undo_redo_sim.py:test_undo_redo_lifecycle, tests/test_visual_mma.py:test_visual_mma_components, tests/test_workspace_profiles_sim.py:test_workspace_profiles_restoration] """ return self.client.get_value(tag) def wait_for_event(self, event_type: str, timeout: int = 5) -> Optional[dict]: """ [C: simulation/sim_execution.py:ExecutionSimulation.run, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout] """ return self.client.wait_for_event(event_type, timeout) def assert_panel_visible(self, panel_tag: str, msg: str = None) -> None: pass def wait_for_element(self, tag: str, timeout: int = 2) -> bool: start = time.time() while time.time() - start < timeout: try: # If we can get_value without error, it's likely there self.client.get_value(tag) return True except: time.sleep(0.1) return False def run_sim(sim_class: type) -> None: """ Helper to run a simulation class standalone. [C: simulation/sim_context.py:module, simulation/sim_execution.py:module, simulation/sim_tools.py:module] """ sim = sim_class() try: sim.setup() sim.run() print(f"\n[SUCCESS] {sim_class.__name__} completed successfully.") except Exception as e: print(f"\n[FAILURE] {sim_class.__name__} failed: {e}") import traceback traceback.print_exc() sys.exit(1) finally: sim.teardown()