test(sim): Add failing simulation test for MMA Step Mode approval

2026-05-02 13:45:08 -04:00
parent db85d60de7
commit 9f67a31b2f
1 changed files with 71 additions and 0 deletions
@@ -0,0 +1,71 @@
+import pytest
+import time
+import sys
+import os
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
+
+from src import api_hook_client
+
+def _poll_mma_status(client: api_hook_client.ApiHookClient, timeout: int, condition, label: str) -> tuple[bool, dict]:
+ """Poll get_mma_status() until condition(status) is True or timeout."""
+ status = {}
+ for i in range(timeout):
+  status = client.get_mma_status() or {}
+  print(f"[SIM][{label}] t={i}s mma={status.get('mma_status')} mode={status.get('mma_step_mode')} tickets={[t.get('status') for t in status.get('active_tickets', [])]}")
+  if condition(status):
+   return True, status
+  time.sleep(1)
+ return False, status
+
+@pytest.mark.integration
+@pytest.mark.timeout(300)
+def test_mma_step_mode_approval_flow(live_gui) -> None:
+ """
+    Verify that we can manually approve a ticket in Step Mode and it proceeds.
+    """
+ client = api_hook_client.ApiHookClient()
+ assert client.wait_for_server(timeout=15), "Hook server did not start"
+
+ # 1. Setup provider and enable Step Mode
+ client.set_value('current_provider', 'gemini_cli')
+ client.set_value('gcli_path', f'"{sys.executable}" "{os.path.abspath("tests/mock_gemini_cli.py")}"')
+ client.click('btn_project_save')
+ client.pause_mma_pipeline()
+ time.sleep(1.0)
+
+ # 2. Create and start a track
+ client.set_value('mma_epic_input', 'TEST APPROVAL')
+ client.click('btn_mma_plan_epic')
+ _poll_mma_status(client, timeout=60, label="wait-proposed", condition=lambda s: bool(s.get('proposed_tracks')))
+ client.click('btn_mma_accept_tracks')
+ _poll_mma_status(client, timeout=30, label="wait-tracks", condition=lambda s: bool(s.get('tracks')))
+ 
+ status = client.get_mma_status()
+ track_id = status['tracks'][0]['id']
+ client.click('btn_mma_load_track', user_data=track_id)
+ time.sleep(0.5)
+ client.click('btn_mma_start_track', user_data=track_id)
+ 
+ # 3. Wait for ticket to be 'ready' (Awaiting Approval)
+ ok, status = _poll_mma_status(client, timeout=20, label="wait-awaiting",
+  condition=lambda s: "Awaiting Approval" in str(s.get('active_tier', '')))
+ assert ok, "Did not reach Awaiting Approval state"
+ 
+ tickets = status.get('active_tickets', [])
+ tid = tickets[0]['id']
+ 
+ # 4. Attempt to approve (THIS SHOULD FAIL OR DO NOTHING CURRENTLY)
+ print(f"[SIM] Attempting to approve ticket {tid}...")
+ # We'll try to use mutate_mma_dag to set status to in_progress
+ # (Note: this uses the bugged '_mutate_dag' endpoint internally if not fixed)
+ res = client.mutate_mma_dag({"ticket_id": tid, "status": "in_progress"})
+ print(f"[SIM] Mutate result: {res}")
+ 
+ # 5. Verify it moved to in_progress
+ ok, status = _poll_mma_status(client, timeout=10, label="verify-in-progress",
+  condition=lambda s: any(t['id'] == tid and t['status'] == 'in_progress' for t in s.get('active_tickets', [])))
+ 
+ assert ok, "Ticket did not move to in_progress after manual approval/mutation"
+ print("[SIM] MMA Step Mode approval flow test PASSED.")