80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""Regression test: _handle_reset_session must clear MMA + RAG state.
|
|
|
|
Added in test_infrastructure_hardening_20260609 followup. Without these
|
|
resets, prior test state pollutes the next session via:
|
|
- mma_tier_usage: stale per-tier model/input/output values cause the
|
|
mma_state_update handler to merge with old data, leaving the new
|
|
push's active_tier effectively lost in the GUI render path.
|
|
- rag_engine / rag_config: the chroma collection from a prior test
|
|
doesn't get rebuilt, causing RAG queries to return nothing.
|
|
"""
|
|
import time
|
|
import pytest
|
|
|
|
from src import api_hook_client
|
|
|
|
|
|
@pytest.mark.clean_baseline
|
|
def test_reset_session_clears_mma_tier_usage(live_gui) -> None:
|
|
"""reset_session() must restore mma_tier_usage to the default (all empty)."""
|
|
client = api_hook_client.ApiHookClient()
|
|
assert client.wait_for_server(timeout=15)
|
|
# Pollute the mma_tier_usage via a mma_state_update push
|
|
client.push_event('mma_state_update', {
|
|
'status': 'running',
|
|
'active_tier': 'Tier 3 (Worker)',
|
|
'tier_usage': {'Tier 1': {'model': 'polluted'}},
|
|
'tickets': []
|
|
})
|
|
time.sleep(0.5)
|
|
# Trigger the reset
|
|
client.reset_session()
|
|
time.sleep(0.5)
|
|
# Verify the polluted entry is gone
|
|
state = client.get_gui_state()
|
|
mma = state.get('mma_state', {})
|
|
tier1 = mma.get('tier_usage', {}).get('Tier 1', {})
|
|
assert tier1.get('model') != 'polluted', (
|
|
f"reset_session() did not clear mma_tier_usage: {tier1!r}"
|
|
)
|
|
|
|
|
|
@pytest.mark.clean_baseline
|
|
def test_reset_session_clears_mma_status(live_gui) -> None:
|
|
"""reset_session() must set mma_status back to 'idle'."""
|
|
client = api_hook_client.ApiHookClient()
|
|
assert client.wait_for_server(timeout=15)
|
|
client.push_event('mma_state_update', {
|
|
'status': 'running',
|
|
'active_tier': 'Tier 1 (Orchestrator)',
|
|
'tier_usage': {},
|
|
'tickets': []
|
|
})
|
|
time.sleep(0.5)
|
|
client.reset_session()
|
|
time.sleep(0.5)
|
|
state = client.get_gui_state()
|
|
assert state.get('mma_status') == 'idle', (
|
|
f"mma_status not reset: {state.get('mma_status')!r}"
|
|
)
|
|
|
|
|
|
@pytest.mark.clean_baseline
|
|
def test_reset_session_clears_active_tier(live_gui) -> None:
|
|
"""reset_session() must set active_tier back to None."""
|
|
client = api_hook_client.ApiHookClient()
|
|
assert client.wait_for_server(timeout=15)
|
|
client.push_event('mma_state_update', {
|
|
'status': 'running',
|
|
'active_tier': 'Tier 2 (Tech Lead)',
|
|
'tier_usage': {},
|
|
'tickets': []
|
|
})
|
|
time.sleep(0.5)
|
|
client.reset_session()
|
|
time.sleep(0.5)
|
|
state = client.get_gui_state()
|
|
assert state.get('active_tier') is None, (
|
|
f"active_tier not reset: {state.get('active_tier')!r}"
|
|
)
|