115 lines
No EOL
3.6 KiB
Python
115 lines
No EOL
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify memory management fixes
|
|
"""
|
|
|
|
import sys
|
|
from memory_manager import MemoryManager, reduce_feature_count
|
|
from hybrid_detector import HybridImageDetector
|
|
|
|
def test_memory_manager():
|
|
"""Test memory manager functionality"""
|
|
print("Testing Memory Manager...")
|
|
|
|
mm = MemoryManager(max_memory_percent=75, max_swap_percent=30)
|
|
|
|
# Test memory usage reporting
|
|
usage = mm.get_memory_usage()
|
|
print(f"Current memory usage: {usage['memory_percent']:.1f}%")
|
|
print(f"Current swap usage: {usage['swap_percent']:.1f}%")
|
|
print(f"Available memory: {usage['memory_available_gb']:.1f} GB")
|
|
|
|
# Test that swap usage doesn't block processing
|
|
print(f"\nTesting is_memory_safe with swap usage {usage['swap_percent']:.1f}%:")
|
|
is_safe = mm.is_memory_safe()
|
|
print(f"Memory safe: {is_safe} (should be True if RAM < 75%, regardless of swap)")
|
|
|
|
# Test concurrent process limiting
|
|
safe_processes = mm.limit_concurrent_processes()
|
|
print(f"Safe concurrent processes: {safe_processes}")
|
|
|
|
print("Memory Manager test completed ✓")
|
|
|
|
def test_hybrid_detector_memory_settings():
|
|
"""Test hybrid detector memory settings"""
|
|
print("\nTesting Hybrid Detector Memory Settings...")
|
|
|
|
try:
|
|
detector = HybridImageDetector(
|
|
panel_threshold=2,
|
|
inlier_threshold=0.65,
|
|
local_workers=4 # Reduced for testing
|
|
)
|
|
|
|
print(f"Memory manager initialized: {detector.memory_manager is not None}")
|
|
print(f"Max memory percent: {detector.memory_manager.max_memory_percent}%")
|
|
print(f"Max swap percent: {detector.memory_manager.max_swap_percent}%")
|
|
|
|
print("Hybrid Detector memory settings test completed ✓")
|
|
|
|
except Exception as e:
|
|
print(f"Error testing hybrid detector: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_feature_reduction():
|
|
"""Test feature reduction functionality"""
|
|
print("\nTesting Feature Reduction...")
|
|
|
|
# Mock features (normally cv2.KeyPoint objects)
|
|
class MockFeature:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
|
|
# Create mock features
|
|
features = [MockFeature(i) for i in range(15000)]
|
|
print(f"Original feature count: {len(features)}")
|
|
|
|
# Test reduction
|
|
reduced = reduce_feature_count(features, max_features=10000)
|
|
print(f"Reduced feature count: {len(reduced)}")
|
|
|
|
# Should keep the best features (highest response values)
|
|
if len(reduced) == 10000:
|
|
print("Feature reduction test completed ✓")
|
|
return True
|
|
else:
|
|
print("Feature reduction test failed ✗")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
print("="*60)
|
|
print("MEMORY MANAGEMENT TEST SUITE")
|
|
print("="*60)
|
|
|
|
success = True
|
|
|
|
# Test 1: Memory Manager
|
|
try:
|
|
test_memory_manager()
|
|
except Exception as e:
|
|
print(f"Memory Manager test failed: {e}")
|
|
success = False
|
|
|
|
# Test 2: Hybrid Detector
|
|
try:
|
|
success &= test_hybrid_detector_memory_settings()
|
|
except Exception as e:
|
|
print(f"Hybrid Detector test failed: {e}")
|
|
success = False
|
|
|
|
# Test 3: Feature Reduction
|
|
try:
|
|
success &= test_feature_reduction()
|
|
except Exception as e:
|
|
print(f"Feature Reduction test failed: {e}")
|
|
success = False
|
|
|
|
print("\n" + "="*60)
|
|
if success:
|
|
print("✓ ALL TESTS PASSED - Memory management is working")
|
|
print("The system should now be protected against memory crashes.")
|
|
else:
|
|
print("✗ SOME TESTS FAILED - Check the errors above")
|
|
print("="*60) |