50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
import pytest
|
|
import time
|
|
from api_hook_client import ApiHookClient
|
|
|
|
def test_comms_volume_stress_performance():
|
|
"""
|
|
Stress test: Inject many comms entries and verify performance doesn't degrade.
|
|
"""
|
|
client = ApiHookClient(base_url="http://127.0.0.1:8999")
|
|
|
|
try:
|
|
# 1. Capture baseline
|
|
baseline = client.get_performance()['performance']
|
|
baseline_ft = baseline.get('last_frame_time_ms', 0.0)
|
|
|
|
# 2. Inject 50 "dummy" comms entries via the session hook
|
|
# Note: In a real app we might need a specific 'inject_comms' hook if we wanted
|
|
# to test the _flush_pending_comms logic specifically, but updating session
|
|
# often triggers similar UI updates or usage recalculations.
|
|
# Actually, let's use post_session to add a bunch of history entries.
|
|
|
|
large_session = []
|
|
for i in range(50):
|
|
large_session.append({"role": "user", "content": f"Stress test entry {i} " * 10})
|
|
|
|
client.post_session(large_session)
|
|
|
|
# Give it a moment to process UI updates if any
|
|
time.sleep(1.0)
|
|
|
|
# 3. Capture stress performance
|
|
stress = client.get_performance()['performance']
|
|
stress_ft = stress.get('last_frame_time_ms', 0.0)
|
|
|
|
print(f"Baseline FT: {baseline_ft:.2f}ms, Stress FT: {stress_ft:.2f}ms")
|
|
|
|
# Requirement: Still under 16.6ms even with 50 new entries
|
|
assert stress_ft < 16.6, f"Stress frame time {stress_ft:.2f}ms exceeds 16.6ms threshold"
|
|
|
|
except Exception as e:
|
|
pytest.fail(f"Stress test failed: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
client = ApiHookClient(base_url="http://127.0.0.1:8999")
|
|
try:
|
|
perf = client.get_performance()
|
|
print(f"Current performance: {perf}")
|
|
except Exception as e:
|
|
print(f"App not running or error: {e}")
|