checkpoint: this is a mess... need to define stricter DSL or system for how the AI devices sims and hookup api for tests.
This commit is contained in:
@@ -17,11 +17,14 @@ def test_mma_complete_lifecycle(live_gui) -> None:
|
||||
client = ApiHookClient()
|
||||
assert client.wait_for_server(timeout=10)
|
||||
|
||||
# 1. Set model to 'mock'.
|
||||
# 1. Set up the mock CLI provider
|
||||
try:
|
||||
client.set_value('current_model', 'mock')
|
||||
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 model to 'mock': {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')
|
||||
@@ -136,19 +139,30 @@ def test_mma_complete_lifecycle(live_gui) -> None:
|
||||
|
||||
# 8. Verify 'active_tier' change and output in 'mma_streams'.
|
||||
streams_found = False
|
||||
for _ in range(30):
|
||||
for _ in range(60): # Give it more time for the worker to spawn and respond
|
||||
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
|
||||
|
||||
# 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 or 'Tier 1' in status.get('mma_streams', {}), "No output found in 'mma_streams'."
|
||||
assert streams_found, "No Tier 3 mock output found in 'mma_streams'."
|
||||
print("MMA complete lifecycle simulation successful.")
|
||||
|
||||
Reference in New Issue
Block a user