145 lines
6.4 KiB
Python
145 lines
6.4 KiB
Python
"""
|
|
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() |