""" 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: 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: 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: return self.client.get_value(tag) def wait_for_event(self, event_type: str, timeout: int = 5) -> Optional[dict]: 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.""" 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()