feat(tests): Migrate manual launchers to live_gui fixture and consolidate visual tests
This commit is contained in:
@@ -42,7 +42,11 @@ files = [
|
|||||||
"tests/test_mma_orchestration_gui.py",
|
"tests/test_mma_orchestration_gui.py",
|
||||||
"tests/test_headless_verification.py",
|
"tests/test_headless_verification.py",
|
||||||
"tests/test_tier4_interceptor.py",
|
"tests/test_tier4_interceptor.py",
|
||||||
"tests/test_conductor_api_hook_integration.py"
|
"tests/test_conductor_api_hook_integration.py",
|
||||||
|
"tests/test_visual_mma.py",
|
||||||
|
"tests/test_visual_orchestration.py",
|
||||||
|
"tests/test_visual_sim_gui_ux.py",
|
||||||
|
"tests/test_visual_sim_mma_v2.py"
|
||||||
]
|
]
|
||||||
|
|
||||||
[categories.integrations]
|
[categories.integrations]
|
||||||
|
|||||||
71
tests/test_visual_mma.py
Normal file
71
tests/test_visual_mma.py
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
import pytest
|
||||||
|
import time
|
||||||
|
from api_hook_client import ApiHookClient
|
||||||
|
|
||||||
|
def test_visual_mma_components(live_gui):
|
||||||
|
"""
|
||||||
|
Refactored visual MMA verification using the live_gui fixture.
|
||||||
|
Ensures the MMA dashboard and tickets are correctly rendered.
|
||||||
|
"""
|
||||||
|
# live_gui is a tuple (process, script_name)
|
||||||
|
_, gui_script = live_gui
|
||||||
|
print(f"Testing visual MMA components on {gui_script}...")
|
||||||
|
|
||||||
|
# 1. Initialize ApiHookClient
|
||||||
|
# The fixture ensures the server is already ready
|
||||||
|
client = ApiHookClient()
|
||||||
|
print("ApiHookClient initialized successfully.")
|
||||||
|
|
||||||
|
# 2. Setup MMA data
|
||||||
|
track_data = {
|
||||||
|
"id": "visual_test_track",
|
||||||
|
"title": "Visual Verification Track",
|
||||||
|
"description": "A track to verify MMA UI components"
|
||||||
|
}
|
||||||
|
tickets_data = [
|
||||||
|
{"id": "TICKET-001", "target_file": "core.py", "status": "todo"},
|
||||||
|
{"id": "TICKET-002", "target_file": "utils.py", "status": "running"},
|
||||||
|
{"id": "TICKET-003", "target_file": "tests.py", "status": "complete"},
|
||||||
|
{"id": "TICKET-004", "target_file": "api.py", "status": "blocked"},
|
||||||
|
{"id": "TICKET-005", "target_file": "gui.py", "status": "paused"},
|
||||||
|
]
|
||||||
|
|
||||||
|
print("\nPushing MMA state update...")
|
||||||
|
payload = {
|
||||||
|
"status": "running",
|
||||||
|
"active_tier": "Tier 3",
|
||||||
|
"track": track_data,
|
||||||
|
"tickets": tickets_data
|
||||||
|
}
|
||||||
|
client.push_event("mma_state_update", payload)
|
||||||
|
print(" - MMA state update pushed.")
|
||||||
|
|
||||||
|
# Use ApiHookClient.wait_for_event if we had a specific event to wait for,
|
||||||
|
# but here we just want to verify state.
|
||||||
|
time.sleep(1) # Small sleep for UI to catch up with event queue
|
||||||
|
|
||||||
|
# 3. Trigger HITL modal
|
||||||
|
print("Pushing 'mma_step_approval' event to trigger HITL modal...")
|
||||||
|
approval_payload = {
|
||||||
|
"ticket_id": "TICKET-002",
|
||||||
|
"payload": "powershell -Command \"Write-Host 'Hello from Tier 3'\""
|
||||||
|
}
|
||||||
|
client.push_event("mma_step_approval", approval_payload)
|
||||||
|
print("mma_step_approval event pushed successfully.")
|
||||||
|
|
||||||
|
# 4. Assertions
|
||||||
|
# We can verify internal state via get_value if hooks are available
|
||||||
|
# For now, we verify the push was successful (it would raise if not)
|
||||||
|
# and we can check some values that should have changed.
|
||||||
|
active_tier = client.get_value("mma_active_tier")
|
||||||
|
assert active_tier == "Tier 3"
|
||||||
|
|
||||||
|
# Verify ticket count if possible
|
||||||
|
# mma_tickets might be a complex object, we'll see if get_value handles it
|
||||||
|
tickets = client.get_value("mma_tickets")
|
||||||
|
if tickets:
|
||||||
|
assert len(tickets) == 5
|
||||||
|
assert tickets[1]['id'] == "TICKET-002"
|
||||||
|
assert tickets[1]['status'] == "running"
|
||||||
|
|
||||||
|
print("Visual MMA component verification PASSED.")
|
||||||
@@ -1,86 +0,0 @@
|
|||||||
import subprocess
|
|
||||||
import time
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
import unittest
|
|
||||||
|
|
||||||
# Calculate project root
|
|
||||||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
|
||||||
if PROJECT_ROOT not in sys.path:
|
|
||||||
sys.path.insert(0, PROJECT_ROOT)
|
|
||||||
|
|
||||||
from api_hook_client import ApiHookClient
|
|
||||||
|
|
||||||
class TestMMAGUIRobust(unittest.TestCase):
|
|
||||||
@classmethod
|
|
||||||
def setUpClass(cls) -> None:
|
|
||||||
# 1. Launch gui_2.py with --enable-test-hooks
|
|
||||||
cls.gui_command = [sys.executable, "gui_2.py", "--enable-test-hooks"]
|
|
||||||
print(f"Launching GUI: {' '.join(cls.gui_command)}")
|
|
||||||
cls.gui_process = subprocess.Popen(
|
|
||||||
cls.gui_command,
|
|
||||||
cwd=PROJECT_ROOT,
|
|
||||||
stdout=subprocess.PIPE,
|
|
||||||
stderr=subprocess.PIPE,
|
|
||||||
text=True
|
|
||||||
)
|
|
||||||
cls.client = ApiHookClient()
|
|
||||||
print("Waiting for GUI to start...")
|
|
||||||
if not cls.client.wait_for_server(timeout=10):
|
|
||||||
cls.gui_process.terminate()
|
|
||||||
raise RuntimeError("GUI failed to start or hook server not responsive.")
|
|
||||||
print("GUI started.")
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def tearDownClass(cls) -> None:
|
|
||||||
if cls.gui_process:
|
|
||||||
cls.gui_process.terminate()
|
|
||||||
cls.gui_process.wait(timeout=5)
|
|
||||||
|
|
||||||
def test_mma_state_ingestion(self) -> None:
|
|
||||||
"""Verify that mma_state_update event correctly updates GUI state."""
|
|
||||||
track_data = {
|
|
||||||
"id": "robust_test_track",
|
|
||||||
"title": "Robust Verification Track",
|
|
||||||
"description": "Verifying internal state ingestion"
|
|
||||||
}
|
|
||||||
tickets_data = [
|
|
||||||
{"id": "T1", "target_file": "file1.py", "status": "todo"},
|
|
||||||
{"id": "T2", "target_file": "file2.py", "status": "running"},
|
|
||||||
{"id": "T3", "target_file": "file3.py", "status": "complete"},
|
|
||||||
]
|
|
||||||
payload = {
|
|
||||||
"status": "active",
|
|
||||||
"active_tier": "Tier 2",
|
|
||||||
"track": track_data,
|
|
||||||
"tickets": tickets_data
|
|
||||||
}
|
|
||||||
print("Pushing mma_state_update...")
|
|
||||||
self.client.push_event("mma_state_update", payload)
|
|
||||||
# Give GUI a moment to process the async task
|
|
||||||
time.sleep(1.0)
|
|
||||||
print("Querying mma_status...")
|
|
||||||
status = self.client.get_mma_status()
|
|
||||||
self.assertEqual(status["mma_status"], "active")
|
|
||||||
self.assertEqual(status["active_tier"], "Tier 2")
|
|
||||||
self.assertEqual(status["active_track"]["id"], "robust_test_track")
|
|
||||||
self.assertEqual(len(status["active_tickets"]), 3)
|
|
||||||
self.assertEqual(status["active_tickets"][2]["status"], "complete")
|
|
||||||
print("MMA state ingestion verified successfully.")
|
|
||||||
|
|
||||||
def test_mma_step_approval_trigger(self) -> None:
|
|
||||||
"""Verify that mma_step_approval event sets the pending approval flag."""
|
|
||||||
payload = {
|
|
||||||
"ticket_id": "T2",
|
|
||||||
"payload": "echo 'Robust Test'"
|
|
||||||
}
|
|
||||||
print("Pushing mma_step_approval...")
|
|
||||||
self.client.push_event("mma_step_approval", payload)
|
|
||||||
time.sleep(1.0)
|
|
||||||
print("Querying mma_status for pending approval...")
|
|
||||||
status = self.client.get_mma_status()
|
|
||||||
self.assertTrue(status["pending_approval"], "GUI did not register pending MMA approval")
|
|
||||||
print("MMA step approval trigger verified successfully.")
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
@@ -1,62 +0,0 @@
|
|||||||
import subprocess
|
|
||||||
import time
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
|
||||||
if PROJECT_ROOT not in sys.path:
|
|
||||||
sys.path.insert(0, PROJECT_ROOT)
|
|
||||||
|
|
||||||
from api_hook_client import ApiHookClient
|
|
||||||
|
|
||||||
def diag_run() -> None:
|
|
||||||
print("Launching GUI for manual inspection + automated hooks...")
|
|
||||||
# Use a log file for GUI output
|
|
||||||
with open("gui_diag.log", "w") as log_file:
|
|
||||||
gui_process = subprocess.Popen(
|
|
||||||
[sys.executable, "gui_2.py", "--enable-test-hooks"],
|
|
||||||
cwd=PROJECT_ROOT,
|
|
||||||
stdout=log_file,
|
|
||||||
stderr=log_file,
|
|
||||||
text=True
|
|
||||||
)
|
|
||||||
client = ApiHookClient()
|
|
||||||
print("Waiting for GUI...")
|
|
||||||
if not client.wait_for_server(timeout=10):
|
|
||||||
print("GUI failed to start.")
|
|
||||||
gui_process.terminate()
|
|
||||||
return
|
|
||||||
# Pushing state
|
|
||||||
track_data = {"id": "diag_track", "title": "Diagnostic Track"}
|
|
||||||
tickets_data = [{"id": f"T{i}", "status": "todo"} for i in range(3)]
|
|
||||||
print("Pushing state update...")
|
|
||||||
client.push_event("mma_state_update", {
|
|
||||||
"status": "active",
|
|
||||||
"active_tier": "Tier 1",
|
|
||||||
"track": track_data,
|
|
||||||
"tickets": tickets_data
|
|
||||||
})
|
|
||||||
time.sleep(2)
|
|
||||||
print("Pushing approval request...")
|
|
||||||
client.push_event("mma_step_approval", {
|
|
||||||
"ticket_id": "T0",
|
|
||||||
"payload": "Get-ChildItem"
|
|
||||||
})
|
|
||||||
print("\nGUI is running. Check 'gui_diag.log' for output.")
|
|
||||||
print("I will now poll mma_status every 2 seconds. Ctrl+C to stop.")
|
|
||||||
try:
|
|
||||||
start_poll = time.time()
|
|
||||||
while time.time() - start_poll < 30:
|
|
||||||
try:
|
|
||||||
status = client.get_mma_status()
|
|
||||||
print(f"[{time.strftime('%H:%M:%S')}] Status: {status.get('mma_status')}, Pending Approval: {status.get('pending_approval')}")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"[{time.strftime('%H:%M:%S')}] Error querying status: {e}")
|
|
||||||
time.sleep(2)
|
|
||||||
except KeyboardInterrupt:
|
|
||||||
print("Stopping...")
|
|
||||||
finally:
|
|
||||||
gui_process.terminate()
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
diag_run()
|
|
||||||
@@ -1,126 +0,0 @@
|
|||||||
import subprocess
|
|
||||||
import time
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
# --- Configuration ---
|
|
||||||
GUI_SCRIPT = 'gui_2.py'
|
|
||||||
TEST_HOOKS_FLAG = '--enable-test-hooks'
|
|
||||||
API_HOOK_CLIENT_MODULE = 'api_hook_client'
|
|
||||||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
|
||||||
|
|
||||||
# Ensure project root is in sys.path to import modules like api_hook_client
|
|
||||||
if PROJECT_ROOT not in sys.path:
|
|
||||||
sys.path.insert(0, PROJECT_ROOT)
|
|
||||||
print(f"Added '{PROJECT_ROOT}' to sys.path for imports.")
|
|
||||||
|
|
||||||
try:
|
|
||||||
from api_hook_client import ApiHookClient
|
|
||||||
except ImportError as e:
|
|
||||||
print(f"Error: Could not import ApiHookClient from '{API_HOOK_CLIENT_MODULE}'.")
|
|
||||||
print(f"Please ensure '{API_HOOK_CLIENT_MODULE}.py' is accessible and '{PROJECT_ROOT}' is correctly added to sys.path.")
|
|
||||||
print(f"Import error: {e}")
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
def run_visual_mma_verification() -> None:
|
|
||||||
print("Starting visual MMA verification test...")
|
|
||||||
# Change current directory to project root
|
|
||||||
original_dir = os.getcwd()
|
|
||||||
if original_dir != PROJECT_ROOT:
|
|
||||||
try:
|
|
||||||
os.chdir(PROJECT_ROOT)
|
|
||||||
print(f"Changed current directory to: {PROJECT_ROOT}")
|
|
||||||
except FileNotFoundError:
|
|
||||||
print(f"Error: Project root directory '{PROJECT_ROOT}' not found.")
|
|
||||||
return
|
|
||||||
# 1. Launch gui_2.py with --enable-test-hooks
|
|
||||||
gui_command = [sys.executable, GUI_SCRIPT, TEST_HOOKS_FLAG]
|
|
||||||
print(f"Launching GUI with command: {' '.join(gui_command)}")
|
|
||||||
try:
|
|
||||||
gui_process = subprocess.Popen(
|
|
||||||
gui_command,
|
|
||||||
stdout=subprocess.PIPE,
|
|
||||||
stderr=subprocess.PIPE,
|
|
||||||
text=True,
|
|
||||||
cwd=PROJECT_ROOT
|
|
||||||
)
|
|
||||||
print(f"GUI process started with PID: {gui_process.pid}")
|
|
||||||
except FileNotFoundError:
|
|
||||||
print(f"Error: Could not find Python executable '{sys.executable}' or script '{GUI_SCRIPT}'.")
|
|
||||||
return
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error starting GUI process: {e}")
|
|
||||||
return
|
|
||||||
# Wait for GUI to start
|
|
||||||
print("Waiting for GUI to initialize and hook server to start (5 seconds)...")
|
|
||||||
time.sleep(5)
|
|
||||||
if gui_process.poll() is not None:
|
|
||||||
print(f"Error: GUI process exited prematurely with return code {gui_process.returncode}.")
|
|
||||||
return
|
|
||||||
# 2. Use ApiHookClient
|
|
||||||
try:
|
|
||||||
client = ApiHookClient()
|
|
||||||
print("ApiHookClient initialized successfully.")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Failed to initialize ApiHookClient. Error: {e}")
|
|
||||||
if gui_process:
|
|
||||||
gui_process.terminate()
|
|
||||||
return
|
|
||||||
# 3. Setup MMA data
|
|
||||||
track_data = {
|
|
||||||
"id": "visual_test_track",
|
|
||||||
"title": "Visual Verification Track",
|
|
||||||
"description": "A track to verify MMA UI components"
|
|
||||||
}
|
|
||||||
tickets_data = [
|
|
||||||
{"id": "TICKET-001", "target_file": "core.py", "status": "todo"},
|
|
||||||
{"id": "TICKET-002", "target_file": "utils.py", "status": "running"},
|
|
||||||
{"id": "TICKET-003", "target_file": "tests.py", "status": "complete"},
|
|
||||||
{"id": "TICKET-004", "target_file": "api.py", "status": "blocked"},
|
|
||||||
{"id": "TICKET-005", "target_file": "gui.py", "status": "paused"},
|
|
||||||
]
|
|
||||||
print("\nPushing MMA state update...")
|
|
||||||
try:
|
|
||||||
payload = {
|
|
||||||
"status": "running",
|
|
||||||
"active_tier": "Tier 3",
|
|
||||||
"track": track_data,
|
|
||||||
"tickets": tickets_data
|
|
||||||
}
|
|
||||||
client.push_event("mma_state_update", payload)
|
|
||||||
print(" - MMA state update pushed.")
|
|
||||||
except Exception as e:
|
|
||||||
print(f" - Warning: Failed to push mma_state_update: {e}")
|
|
||||||
time.sleep(3)
|
|
||||||
print("Pushing 'mma_step_approval' event to trigger HITL modal...")
|
|
||||||
try:
|
|
||||||
approval_payload = {
|
|
||||||
"ticket_id": "TICKET-002",
|
|
||||||
"payload": "powershell -Command \"Write-Host 'Hello from Tier 3'\""
|
|
||||||
}
|
|
||||||
client.push_event("mma_step_approval", approval_payload)
|
|
||||||
print("mma_step_approval event pushed successfully.")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error pushing mma_step_approval event: {e}")
|
|
||||||
# 5. Provide clear print statements for manual verification
|
|
||||||
print("\n--- Manual Verification Instructions ---")
|
|
||||||
print("Please visually inspect the running GUI application:")
|
|
||||||
print("1. MMA Dashboard: Ensure the 'MMA Dashboard' panel is visible and active.")
|
|
||||||
print("2. Ticket Queue: Verify the 'Ticket Queue' section displays all 5 tickets with correct statuses.")
|
|
||||||
print("3. Progress Bar: Check that the progress bar correctly reflects the completed/total tickets.")
|
|
||||||
print("4. Approval Modal: Confirm that an 'MMA Step Approval' modal has appeared.")
|
|
||||||
print("\n--------------------------------------")
|
|
||||||
print("The test script has finished its automated actions.")
|
|
||||||
print("The GUI application is still running. Press Enter to exit.")
|
|
||||||
try:
|
|
||||||
input()
|
|
||||||
except EOFError:
|
|
||||||
pass
|
|
||||||
print("\nStopping GUI process...")
|
|
||||||
if gui_process:
|
|
||||||
gui_process.terminate()
|
|
||||||
gui_process.wait(timeout=5)
|
|
||||||
print("Visual MMA verification test script finished.")
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
run_visual_mma_verification()
|
|
||||||
Reference in New Issue
Block a user