manual_slop/tests/visual_sim_mma_v2.py

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)
        # Prevent polluting the real project directory with test tracks
        client.set_value('files_base_dir', 'tests/temp_workspace')
        client.click('btn_project_save')
        time.sleep(1)
    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_mma_spawn_approval') is True:
            print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
            client.click('btn_approve_spawn')
        elif status and status.get('pending_mma_step_approval') is True:
            print('[SIM] MMA step approval required. Clicking btn_approve_mma_step...')
            client.click('btn_approve_mma_step')
        elif status and status.get('pending_tool_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_mma_spawn_approval') is True:
            client.click('btn_approve_spawn')
        elif status and status.get('pending_mma_step_approval') is True:
            client.click('btn_approve_mma_step')
        elif status and status.get('pending_tool_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_mma_spawn_approval') is True:
            print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
            client.click('btn_approve_spawn')
        elif status and status.get('pending_mma_step_approval') is True:
            print('[SIM] MMA step approval required. Clicking btn_approve_mma_step...')
            client.click('btn_approve_mma_step')
        elif status and status.get('pending_tool_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. Poll for MMA status 'running' or 'done' (already started by Accept Tracks).
    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_mma_spawn_approval') is True:
            print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
            client.click('btn_approve_spawn')
        elif status and status.get('pending_mma_step_approval') is True:
            print('[SIM] MMA step approval required. Clicking btn_approve_mma_step...')
            client.click('btn_approve_mma_step')
        elif status and status.get('pending_tool_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_mma_spawn_approval') is True:
            print('[SIM] Worker spawn required. Clicking btn_approve_spawn...')
            client.click('btn_approve_spawn')
        elif status and status.get('pending_mma_step_approval') is True:
            print('[SIM] MMA step approval required. Clicking btn_approve_mma_step...')
            client.click('btn_approve_mma_step')
        elif status and status.get('pending_tool_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.")