chore(mma): Clean up mma_exec.py and robustify visual simulation mocking
This commit is contained in:
@@ -40,3 +40,27 @@ fetch_url = true
|
||||
epic = "Develop a new feature"
|
||||
active_track_id = ""
|
||||
tracks = []
|
||||
|
||||
[mma.active_track]
|
||||
id = "track_024370f1b453"
|
||||
description = "Mock Goal 1"
|
||||
|
||||
[[mma.active_track.tickets]]
|
||||
id = "mock-ticket-1"
|
||||
description = "Mock Ticket 1"
|
||||
status = "todo"
|
||||
assigned_to = "unassigned"
|
||||
context_requirements = []
|
||||
depends_on = []
|
||||
step_mode = false
|
||||
|
||||
[[mma.active_track.tickets]]
|
||||
id = "mock-ticket-2"
|
||||
description = "Mock Ticket 2"
|
||||
status = "todo"
|
||||
assigned_to = "unassigned"
|
||||
context_requirements = []
|
||||
depends_on = [
|
||||
"mock-ticket-1",
|
||||
]
|
||||
step_mode = false
|
||||
|
||||
@@ -10,97 +10,145 @@ 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)
|
||||
client = ApiHookClient()
|
||||
assert client.wait_for_server(timeout=10)
|
||||
|
||||
# 1. Set model to 'mock'.
|
||||
try:
|
||||
client.set_value('current_model', 'mock')
|
||||
except Exception as e:
|
||||
pytest.fail(f"Failed to set model to 'mock': {e}")
|
||||
# 1. Set model to 'mock'.
|
||||
try:
|
||||
client.set_value('current_model', 'mock')
|
||||
except Exception as e:
|
||||
pytest.fail(f"Failed to set model to 'mock': {e}")
|
||||
|
||||
# 2. Enter epic and click 'Plan Epic'.
|
||||
client.set_value('mma_epic_input', 'Develop a new feature')
|
||||
client.click('btn_mma_plan_epic')
|
||||
# 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."
|
||||
# 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(5) # Add delay to ensure background thread processes track refresh
|
||||
# 4. Click 'Accept' to start tracks.
|
||||
client.click('btn_mma_accept_tracks')
|
||||
time.sleep(2)
|
||||
|
||||
# 5. Wait for 'tracks' list to populate.
|
||||
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:
|
||||
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('tracks') and len(status['tracks']) > 0:
|
||||
tracks_populated = True
|
||||
break
|
||||
time.sleep(1)
|
||||
assert tracks_populated, "Failed to populate tracks list after accepting proposed tracks."
|
||||
# 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."
|
||||
# 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}")
|
||||
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)
|
||||
# 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."
|
||||
# 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')
|
||||
|
||||
print(f"Successfully loaded and verified track ID: {track_id_to_load} with active tickets.")
|
||||
# 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(30):
|
||||
status = client.get_mma_status()
|
||||
streams = status.get('mma_streams', {})
|
||||
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())}")
|
||||
streams_found = True
|
||||
break
|
||||
# Keep approving if needed
|
||||
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')
|
||||
time.sleep(1)
|
||||
|
||||
assert streams_found or 'Tier 1' in status.get('mma_streams', {}), "No output found in 'mma_streams'."
|
||||
print("MMA complete lifecycle simulation successful.")
|
||||
|
||||
Reference in New Issue
Block a user