62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Test outlier display in UI"""
|
|
|
|
import sys
|
|
sys.path.append('.')
|
|
import time
|
|
|
|
from analyzer.analysis import EthernetAnalyzer
|
|
from analyzer.utils import PCAPLoader
|
|
from analyzer.analysis.background_analyzer import BackgroundAnalyzer
|
|
|
|
def test_outlier_processing(pcap_file, src_ip="192.168.4.89"):
|
|
"""Test outlier processing through background analyzer"""
|
|
|
|
# Create analyzer
|
|
analyzer = EthernetAnalyzer(outlier_threshold_sigma=3.0)
|
|
|
|
# Create background analyzer
|
|
bg_analyzer = BackgroundAnalyzer(analyzer, num_threads=4)
|
|
|
|
print("Starting background parsing...")
|
|
bg_analyzer.start_parsing(pcap_file)
|
|
|
|
# Wait for processing to complete
|
|
while bg_analyzer.is_parsing:
|
|
print(f"\rProcessing... packets: {bg_analyzer.processed_packets}", end="")
|
|
time.sleep(0.1)
|
|
|
|
print("\n\nProcessing complete!")
|
|
|
|
# Find the specific flow
|
|
target_flow = None
|
|
for flow_key, flow in analyzer.flows.items():
|
|
if flow.src_ip == src_ip:
|
|
target_flow = flow
|
|
print(f"\nFound flow: {flow.src_ip}:{flow.src_port} -> {flow.dst_ip}:{flow.dst_port}")
|
|
break
|
|
|
|
if not target_flow:
|
|
print(f"Flow from {src_ip} not found!")
|
|
return
|
|
|
|
print(f"Total packets in flow: {target_flow.frame_count}")
|
|
print(f"Average inter-arrival: {target_flow.avg_inter_arrival * 1000:.3f} ms")
|
|
print(f"Std deviation: {target_flow.std_inter_arrival * 1000:.3f} ms")
|
|
print(f"Total outliers detected: {len(target_flow.outlier_frames)}")
|
|
print(f"Outlier frames: {target_flow.outlier_frames}")
|
|
|
|
# Check if outliers match expected
|
|
expected_outliers = [1576, 1582, 1634, 1640]
|
|
if set(target_flow.outlier_frames) == set(expected_outliers):
|
|
print("\n✅ SUCCESS: All expected outliers detected!")
|
|
else:
|
|
print("\n❌ FAILURE: Outlier mismatch")
|
|
print(f"Expected: {expected_outliers}")
|
|
print(f"Got: {target_flow.outlier_frames}")
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1:
|
|
test_outlier_processing(sys.argv[1])
|
|
else:
|
|
test_outlier_processing("1 PTPGM.pcapng") |