169 lines
7.5 KiB
Python
169 lines
7.5 KiB
Python
import pytest
|
|
import time
|
|
import sys
|
|
import os
|
|
|
|
# Ensure project root is in path
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from api_hook_client import ApiHookClient
|
|
|
|
@pytest.mark.integration
|
|
def test_mma_complete_lifecycle(live_gui) -> None:
|
|
"""
|
|
Tests the entire MMA lifecycle from epic planning to track loading and ticket verification
|
|
in a single test case to avoid state dependency issues between separate test functions.
|
|
"""
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=10)
|
|
|
|
# 1. Set up the mock CLI provider
|
|
try:
|
|
client.set_value('current_provider', 'gemini_cli')
|
|
# Point the CLI adapter to our mock script
|
|
mock_cli_path = f'{sys.executable} {os.path.abspath("tests/mock_gemini_cli.py")}'
|
|
client.set_value('gcli_path', mock_cli_path)
|
|
except Exception as e:
|
|
pytest.fail(f"Failed to set up mock provider: {e}")
|
|
|
|
# 2. Enter epic and click 'Plan Epic'.
|
|
client.set_value('mma_epic_input', 'Develop a new feature')
|
|
client.click('btn_mma_plan_epic')
|
|
|
|
# 3. Wait for 'proposed_tracks'.
|
|
proposed_tracks_found = False
|
|
for _ in range(60): # Poll for up to 60 seconds
|
|
status = client.get_mma_status()
|
|
print(f"Polling status: {status}")
|
|
print(f"Polling ai_status: {status.get('ai_status', 'N/A')}")
|
|
if status and status.get('pending_spawn') is True:
|
|
print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
|
|
client.click('btn_approve_spawn')
|
|
elif status and status.get('pending_approval') is True:
|
|
print('[SIM] Tool approval required. Clicking btn_approve_tool...')
|
|
client.click('btn_approve_tool')
|
|
if status and status.get('proposed_tracks') and len(status['proposed_tracks']) > 0:
|
|
proposed_tracks_found = True
|
|
break
|
|
time.sleep(1)
|
|
assert proposed_tracks_found, "Failed to find proposed tracks after planning epic."
|
|
|
|
# 4. Click 'Accept' to start tracks.
|
|
client.click('btn_mma_accept_tracks')
|
|
time.sleep(2)
|
|
|
|
# 5. Wait for 'tracks' list to populate with our mock tracks.
|
|
tracks_populated = False
|
|
for _ in range(30): # Poll for up to 30 seconds
|
|
status = client.get_mma_status()
|
|
if status and status.get('pending_spawn') is True:
|
|
client.click('btn_approve_spawn')
|
|
elif status and status.get('pending_approval') is True:
|
|
client.click('btn_approve_tool')
|
|
|
|
tracks = status.get('tracks', [])
|
|
if any('Mock Goal 1' in t.get('title', '') for t in tracks):
|
|
tracks_populated = True
|
|
break
|
|
time.sleep(1)
|
|
assert tracks_populated, "Failed to find 'Mock Goal 1' in tracks list after acceptance."
|
|
|
|
# 6. Verify that one of the new tracks can be loaded and its tickets appear in 'active_tickets'.
|
|
status_after_tracks = client.get_mma_status()
|
|
assert status_after_tracks is not None, "Failed to get MMA status after tracks populated."
|
|
tracks_list = status_after_tracks.get('tracks')
|
|
assert tracks_list is not None and len(tracks_list) > 0, "Tracks list is empty or not found."
|
|
|
|
track_id_to_load = None
|
|
for track in tracks_list:
|
|
if 'Mock Goal 1' in track.get('title', ''):
|
|
track_id_to_load = track['id']
|
|
break
|
|
assert track_id_to_load is not None, "Could not find a track with 'Mock Goal 1' in its title."
|
|
print(f"Attempting to load track with ID: {track_id_to_load}")
|
|
|
|
# Load the first track
|
|
client.click('btn_mma_load_track', user_data=track_id_to_load)
|
|
|
|
# Poll until 'active_track' is not None and 'active_tickets' are present
|
|
active_track_and_tickets_found = False
|
|
for _ in range(60): # Poll for up to 60 seconds
|
|
status = client.get_mma_status()
|
|
print(f"Polling load status: {status}")
|
|
if status and status.get('pending_spawn') is True:
|
|
print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
|
|
client.click('btn_approve_spawn')
|
|
elif status and status.get('pending_approval') is True:
|
|
print('[SIM] Tool approval required. Clicking btn_approve_tool...')
|
|
client.click('btn_approve_tool')
|
|
|
|
# Updated condition to correctly check active_track ID or value
|
|
active_track = status.get('active_track')
|
|
if status and ( (isinstance(active_track, dict) and active_track.get('id') == track_id_to_load) or (active_track == track_id_to_load) ) and \
|
|
'active_tickets' in status and len(status['active_tickets']) > 0:
|
|
active_track_and_tickets_found = True
|
|
break
|
|
time.sleep(1)
|
|
assert active_track_and_tickets_found, f"Timed out waiting for track {track_id_to_load} to load and populate active tickets."
|
|
|
|
print(f"Successfully loaded and verified track ID: {track_id_to_load} with active tickets.")
|
|
|
|
# 7. Start the MMA track and poll for its status.
|
|
print(f"Starting track {track_id_to_load}...")
|
|
client.click('btn_mma_start_track', user_data=track_id_to_load)
|
|
|
|
mma_running = False
|
|
for _ in range(120): # Poll for up to 120 seconds
|
|
status = client.get_mma_status()
|
|
print(f"Polling MMA status for 'running': {status.get('mma_status')}")
|
|
|
|
# Handle pending states during the run
|
|
if status and status.get('pending_spawn') is True:
|
|
print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
|
|
client.click('btn_approve_spawn')
|
|
elif status and status.get('pending_approval') is True:
|
|
print('[SIM] Tool approval required. Clicking btn_approve_tool...')
|
|
client.click('btn_approve_tool')
|
|
|
|
# Check if MMA is running
|
|
if status and status.get('mma_status') == 'running':
|
|
mma_running = True
|
|
break
|
|
# Also check if it's already finished or error
|
|
if status and status.get('mma_status') in ['done', 'error']:
|
|
break
|
|
time.sleep(1)
|
|
assert mma_running or (status and status.get('mma_status') == 'done'), f"Timed out waiting for MMA status to become 'running' for track {track_id_to_load}."
|
|
|
|
print(f"MMA status is: {status.get('mma_status')}")
|
|
|
|
# 8. Verify 'active_tier' change and output in 'mma_streams'.
|
|
streams_found = False
|
|
for _ in range(60): # Give it more time for the worker to spawn and respond
|
|
status = client.get_mma_status()
|
|
|
|
# Handle approvals if they pop up during worker execution
|
|
if status and status.get('pending_spawn') is True:
|
|
print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
|
|
client.click('btn_approve_spawn')
|
|
elif status and status.get('pending_approval') is True:
|
|
print('[SIM] Tool approval required. Clicking btn_approve_tool...')
|
|
client.click('btn_approve_tool')
|
|
|
|
streams = status.get('mma_streams', {})
|
|
print(f"Polling streams: {list(streams.keys())}")
|
|
|
|
if streams and any("Tier 3" in k for k in streams.keys()):
|
|
print(f"[SIM] Found Tier 3 worker output in streams: {list(streams.keys())}")
|
|
# Check for our specific mock content
|
|
tier3_key = [k for k in streams.keys() if "Tier 3" in k][0]
|
|
if "SUCCESS: Mock Tier 3 worker" in streams[tier3_key]:
|
|
print("[SIM] Verified mock worker output content.")
|
|
streams_found = True
|
|
break
|
|
|
|
time.sleep(1)
|
|
|
|
assert streams_found, "No Tier 3 mock output found in 'mma_streams'."
|
|
print("MMA complete lifecycle simulation successful.")
|