Ed_
d1ce0eaaeb
- Add cost tracking with new cost_tracker.py module - Enhance Track Proposal modal with editable titles and goals - Add Conductor Setup summary and New Track creation form to MMA Dashboard - Implement Task DAG editing (add/delete tickets) and track-scoped discussion - Add visual polish: color-coded statuses, tinted progress bars, and node indicators - Support live worker streaming from AI providers to GUI panels - Fix numerous integration test regressions and stabilize headless service
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
import pytest
|
|
import time
|
|
import sys
|
|
import os
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from api_hook_client import ApiHookClient
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.timeout(60)
|
|
def test_gui_ux_event_routing(live_gui) -> None:
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=15), "Hook server did not start"
|
|
|
|
# ------------------------------------------------------------------
|
|
# 1. Verify Streaming Event Routing
|
|
# ------------------------------------------------------------------
|
|
print("[SIM] Testing Streaming Event Routing...")
|
|
stream_id = "Tier 3 (Worker): T-SIM-001"
|
|
|
|
# We use push_event which POSTs to /api/gui with action=mma_stream_append
|
|
# As defined in App._process_pending_gui_tasks
|
|
client.push_event('mma_stream_append', {'stream_id': stream_id, 'text': 'Hello '})
|
|
time.sleep(0.5)
|
|
client.push_event('mma_stream_append', {'stream_id': stream_id, 'text': 'World!'})
|
|
time.sleep(1.0)
|
|
|
|
status = client.get_mma_status()
|
|
streams = status.get('mma_streams', {})
|
|
assert streams.get(stream_id) == 'Hello World!', f"Streaming failed: {streams.get(stream_id)}"
|
|
print("[SIM] Streaming event routing verified.")
|
|
|
|
# ------------------------------------------------------------------
|
|
# 2. Verify State Update (Usage/Cost) Routing
|
|
# ------------------------------------------------------------------
|
|
print("[SIM] Testing State Update Routing...")
|
|
usage = {
|
|
"Tier 1": {"input": 1000, "output": 500, "model": "gemini-3.1-pro-preview"},
|
|
"Tier 2": {"input": 2000, "output": 1000, "model": "gemini-3-flash-preview"}
|
|
}
|
|
|
|
client.push_event('mma_state_update', {
|
|
'status': 'simulating',
|
|
'tier_usage': usage,
|
|
'tickets': []
|
|
})
|
|
time.sleep(1.0)
|
|
|
|
status = client.get_mma_status()
|
|
assert status.get('mma_status') == 'simulating'
|
|
# The app merges or replaces usage. Let's check what we got back.
|
|
received_usage = status.get('mma_tier_usage', {})
|
|
assert received_usage.get('Tier 1', {}).get('input') == 1000
|
|
assert received_usage.get('Tier 2', {}).get('model') == 'gemini-3-flash-preview'
|
|
print("[SIM] State update routing verified.")
|
|
|
|
if __name__ == "__main__":
|
|
pass
|