141 lines
4.7 KiB
Python
141 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify the DuplicateIds error fix
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Add analyzer to path
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from analyzer.analysis.core import EthernetAnalyzer
|
|
from analyzer.tui.textual.widgets.filtered_flow_view import FilteredFlowView
|
|
from analyzer.models.flow_stats import FlowStats, FrameTypeStats
|
|
|
|
|
|
def test_duplicate_ids_fix():
|
|
"""Test that refresh_frame_types doesn't create duplicate IDs"""
|
|
print("Testing DuplicateIds error fix...")
|
|
|
|
# Create analyzer with mock data
|
|
analyzer = EthernetAnalyzer()
|
|
|
|
# Add flows with frame types that change over time (simulating parsing)
|
|
flow1 = FlowStats(src_ip="192.168.1.1", dst_ip="192.168.1.2")
|
|
flow1.frame_types["PTP-Signaling"] = FrameTypeStats("PTP-Signaling", count=100)
|
|
flow1.frame_types["UDP"] = FrameTypeStats("UDP", count=50)
|
|
analyzer.flows["flow1"] = flow1
|
|
|
|
flow_view = FilteredFlowView(analyzer)
|
|
|
|
print("✅ FilteredFlowView created successfully")
|
|
|
|
# Test throttling mechanism
|
|
print("Testing refresh throttling...")
|
|
start_time = time.time()
|
|
|
|
# Call refresh multiple times rapidly
|
|
for i in range(5):
|
|
flow_view.refresh_frame_types()
|
|
time.sleep(0.1) # Small delay
|
|
|
|
end_time = time.time()
|
|
elapsed = end_time - start_time
|
|
|
|
if hasattr(flow_view, '_last_refresh_time'):
|
|
print("✅ Refresh throttling mechanism in place")
|
|
else:
|
|
print("❌ Refresh throttling mechanism missing")
|
|
return False
|
|
|
|
# Test that we can refresh without errors
|
|
print("Testing multiple refresh calls...")
|
|
try:
|
|
# Simulate changes that would trigger refresh
|
|
flow2 = FlowStats(src_ip="192.168.1.3", dst_ip="192.168.1.4")
|
|
flow2.frame_types["PTP-Signaling"] = FrameTypeStats("PTP-Signaling", count=200)
|
|
flow2.frame_types["CH10-Data"] = FrameTypeStats("CH10-Data", count=500)
|
|
analyzer.flows["flow2"] = flow2
|
|
|
|
# Reset throttling to allow refresh
|
|
flow_view._last_refresh_time = 0
|
|
|
|
# This should not cause DuplicateIds error
|
|
flow_view.refresh_frame_types()
|
|
print("✅ First refresh completed without error")
|
|
|
|
# Add more data and refresh again
|
|
flow3 = FlowStats(src_ip="192.168.1.5", dst_ip="192.168.1.6")
|
|
flow3.frame_types["PTP-Signaling"] = FrameTypeStats("PTP-Signaling", count=300)
|
|
analyzer.flows["flow3"] = flow3
|
|
|
|
# Reset throttling again
|
|
flow_view._last_refresh_time = 0
|
|
|
|
# Second refresh should also work
|
|
flow_view.refresh_frame_types()
|
|
print("✅ Second refresh completed without error")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Refresh failed with error: {e}")
|
|
return False
|
|
|
|
# Test frame type detection
|
|
frame_types = flow_view._get_all_frame_types()
|
|
expected_types = {"PTP-Signaling", "UDP", "CH10-Data"}
|
|
|
|
if expected_types.issubset(set(frame_types.keys())):
|
|
print("✅ Frame type detection working correctly")
|
|
else:
|
|
print(f"❌ Frame type detection failed. Expected {expected_types}, got {set(frame_types.keys())}")
|
|
return False
|
|
|
|
print("✅ All DuplicateIds fix tests passed")
|
|
return True
|
|
|
|
|
|
def test_error_handling():
|
|
"""Test error handling in button creation"""
|
|
print("\nTesting error handling...")
|
|
|
|
analyzer = EthernetAnalyzer()
|
|
flow_view = FilteredFlowView(analyzer)
|
|
|
|
# Test that empty frame types don't cause errors
|
|
try:
|
|
flow_view.refresh_frame_types()
|
|
print("✅ Empty frame types handled gracefully")
|
|
except Exception as e:
|
|
print(f"❌ Empty frame types caused error: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("StreamLens DuplicateIds Fix Test")
|
|
print("=" * 40)
|
|
|
|
try:
|
|
success1 = test_duplicate_ids_fix()
|
|
success2 = test_error_handling()
|
|
|
|
if success1 and success2:
|
|
print(f"\n✅ All DuplicateIds fixes verified!")
|
|
print(f"\n📊 Summary of Fixes:")
|
|
print(f" • Throttling: Only refresh buttons once per second")
|
|
print(f" • Safe removal: Use list() to avoid iteration issues")
|
|
print(f" • Error handling: Try/catch around widget operations")
|
|
print(f" • Duplicate checking: Remove existing widgets before creating")
|
|
print(f" • Race condition prevention: Multiple safety mechanisms")
|
|
else:
|
|
print(f"\n❌ Some tests failed")
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed with error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1) |